在 Airflow 的操作员参数中将 jinja 模板格式化为 INT [英] Format jinja template as INT in operator parameter in Airflow
问题描述
我正在尝试将 jinja 模板参数格式化为整数,以便我可以将其传递给需要 INT(可以是自定义或 PythonOperator)的运算符,但我无法做到.
I'm trying to format a jinja template parameter as an integer so I can pass it to an operator which expects INT (could be custom or PythonOperator) and I'm not able to.
请参阅下面的示例 DAG.我正在使用内置的 Jinja 过滤器 |int
但这不起作用 - 类型仍然是
See sample DAG below. I'm using the built-in Jinja filter | int
but that's not working - the type remains <class 'str'>
我还是 Airflow 的新手,但根据我对 Jinja/Airflow 作品的了解,我认为这是不可能的.我看到了两种主要的解决方法:
I'm still new with Airflow but I don't think this is possible based on what I've read about Jinja/Airflow works. I see two main workarounds:
- 将运算符参数更改为期望字符串并处理下面的转换.
在 单独 PythonOperator 中处理此转换,该 PythonOperator 将字符串转换为 int 并使用 xcom/task 上下文导出该字符串.(我认为这会奏效,但不确定)
- Change the operator parameter to expect string and handle the conversion underneath.
Handle this conversion in a separate PythonOperator which converts the string to an int and export that using xcom/task context. (I think this will work but not sure)
请告诉我任何其他解决方法
Please let me know of any other workarounds
def greet(mystr):
print (mystr)
print(type(mystr))
default_args = {
'owner': 'airflow',
'start_date': days_ago(2)
}
dag = DAG(
'template_dag',
default_args=default_args,
description='template',
schedule_interval='0 13 * * *'
)
with dag:
# foo = "{{ var.value.my_custom_var | int }}" # from variable
foo = "{{ execution_date.int_timestamp | int }}" # built in macro
# could be MyCustomOperator
opr_greet = PythonOperator(task_id='greet',
python_callable=greet,
op_kwargs={'mystr': foo}
)
opr_greet
气流 1.10.11
推荐答案
更新答案:
从 Airflow 2.1 开始,您可以将 render_template_as_native_obj=True
传递给 dag,Airflow 将返回 Python 类型(dict、int 等)而不是字符串.请参阅此拉取请求
As of Airflow 2.1, you can pass render_template_as_native_obj=True
to the dag and Airflow will return the Python type (dict, int, etc) instead of string. See this pull request
dag = DAG(
dag_id="example_template_as_python_object",
schedule_interval=None,
start_date=days_ago(2),
render_template_as_native_obj=True,
)
以前版本的旧答案:
我发现了一个提供最佳解决方法的相关问题,IMO.
I found a related question that provides the best workaround, IMO.
诀窍是使用 PythonOperator,在那里进行数据类型转换,然后使用参数调用主运算符.下面是将 json 字符串转换为 dict 对象的示例.同样可以申请string转int等
The trick is to use a PythonOperator, do the datatype conversion there and then call the main operator with the parameter. Below is an example in converting a json string to a dict object. Same can apply for converting string to int, etc.
def my_func(ds, **kwargs):
ti = kwargs['ti']
body = ti.xcom_pull(task_ids='privious_task_id')
import_body= json.loads(body)
op = CloudSqlInstanceImportOperator(
project_id=GCP_PROJECT_ID,
body= import_body,
instance=INSTANCE_NAME,
gcp_conn_id='postgres_default',
task_id='sql_import_task',
validate_body=True,
)
op.execute()
p = PythonOperator(task_id='python_task', python_callable=my_func)
这篇关于在 Airflow 的操作员参数中将 jinja 模板格式化为 INT的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!