使用PythonOperator的模板文件的气流

运行的结果显示table正确模板化为字符串,但其他人没有提取任何文件进行模板化.

dlamblin$ airflow test test_python_operator_template_dag test_python_operator_template 2017-01-18
[2017-01-18 23:58:06,698] {__init__.py:36} INFO - Using executor SequentialExecutor
[2017-01-18 23:58:07,342] {models.py:154} INFO - Filling up the DagBag from /Users/daniellamblin/airflow/dags
[2017-01-18 23:58:07,620] {models.py:1196} INFO - 
--------------------------------------------------------------------------------
Starting attempt 1 of 1
--------------------------------------------------------------------------------

[2017-01-18 23:58:07,620] {models.py:1219} INFO - Executing  on 2017-01-18 00:00:00
'2017-01-18'
{   u'END_DATE': '2017-01-18',
    u'conf': ,
    u'dag': ,
    u'dag_run': None,
    u'ds_nodash': u'20170118',
    u'end_date': '2017-01-18',
    u'execution_date': datetime.datetime(2017, 1, 18, 0, 0),
    u'latest_date': '2017-01-18',
    u'macros': ,
    u'params': {   'condition_param': True,
                   'message': 'Hello World',
                   'table': 'TEMP_TABLE'},
    u'run_id': None,
    u'tables': None,
    u'task': ,
    u'task_instance': ,
    u'task_instance_key_str': u'test_python_operator_template_dag__test_python_operator_template__20170118',
    'templates_dict': {   'file1': u'pyoptemplate.sql',
                          'file2': u'pyoptemplate.t',
                          'pyoptemplate': u'',
                          'pyoptemplate.sql': u'',
                          'sql': u'pyoptemplate',
                          'table': u'TEMP_TABLE'},
    u'test_mode': True,
    u'ti': ,
    u'tomorrow_ds': '2017-01-19',
    u'tomorrow_ds_nodash': u'20170119',
    u'ts': '2017-01-18T00:00:00',
    u'ts_nodash': u'20170118T000000',
    u'yesterday_ds': '2017-01-17',
    u'yesterday_ds_nodash': u'20170117'}
[2017-01-18 23:58:07,634] {python_operator.py:67} INFO - Done. Returned value was: None

Ardan.. 15

从Airflow 1.8开始,PythonOperator替换其template_ext字段的方式__init__不起作用.任务只检查template_ext上__class__.要创建一个拾取SQL模板文件的PythonOperator,您只需执行以下操作:

class SQLTemplatedPythonOperator(PythonOperator):
    template_ext = ('.sql',)

然后在运行时从任务中访问SQL:

SQLTemplatedPythonOperator(
    templates_dict={'query': 'my_template.sql'},
    params={'my_var': 'my_value'},
    python_callable=my_func,
    provide_context=True,
)

def my_func(**context):
    context['templates_dict']['query']

小智.. 12

最近我遇到了同样的问题并最终解决了它.@Ardan的解决方案是正确的,但只想重复一个更完整的答案,并详细介绍Airflow如何为新手工作.

当然,你首先需要其中一个:

from airflow.operators.python_operator import PythonOperator

class SQLTemplatedPythonOperator(PythonOperator):

    # somehow ('.sql',) doesn't work but tuple of two works...
    template_ext = ('.sql','.abcdefg')

假设您有一个如下所示的sql模板文件:

# stored at path: $AIRFLOW_HOME/sql/some.sql
select {{some_params}} from my_table;

首先确保将文件夹添加到dag参数的搜索路径中.

不要将template_searchpath传递给args,然后将args传递给DAG!它不起作用.

dag = DAG(
    dag_id= "some_name",
    default_args=args,
    schedule_interval="@once",
    template_searchpath='/Users/your_name/some_path/airflow_home/sql'
)

那么您的操作员电话将是

SQLTemplatedPythonOperator(
        templates_dict={'query': 'some.sql'},
        op_kwargs={"args_directly_passed_to_your_function": "some_value"},
        task_id='dummy',
        params={"some_params":"some_value"},
        python_callable=your_func,
        provide_context=True,
        dag=dag,
    )

你的功能将是:

def your_func(args_directly_passed_to_your_function=None):
    query = context['templates_dict']['query']
    dome_some_thing(query)

一些解释:

Airflow使用上下文中的值来呈现模板.要手动将其添加到上下文中,您可以使用上面的params字段.

PythonOperator不再从template_ext字段获取模板文件扩展名,就像@Ardan提到的那样.源代码在这里.它只需要从self .__ class __.template_ext扩展.

气流循环遍历template_dict字段,如果value.endswith(file_extension)== True,则呈现模板.

Will Fitzger.. 9

我认为这不可能.但以下解决方法可能会有所帮助:

def templated_function(ds, **kwargs):
    kwargs['ds'] = ds                                # put ds into 'context'
    task = kwargs['task']                            # get handle on task
    templ = open(kwargs['templates_dict']['file1']).read() # get template
    sql = task.render_template('', tmpl, kwargs)           # render it
    pp.pprint(sql)

但是,我会喜欢更好的解决方案!

从Airflow 1.8开始,PythonOperator替换其template_ext字段的方式__init__不起作用.任务只检查template_ext上__class__.要创建一个拾取SQL模板文件的PythonOperator,您只需执行以下操作:

class SQLTemplatedPythonOperator(PythonOperator):
    template_ext = ('.sql',)

然后在运行时从任务中访问SQL:

SQLTemplatedPythonOperator(
    templates_dict={'query': 'my_template.sql'},
    params={'my_var': 'my_value'},
    python_callable=my_func,
    provide_context=True,
)

def my_func(**context):
    context['templates_dict']['query']

最近我遇到了同样的问题并最终解决了它.@Ardan的解决方案是正确的,但只想重复一个更完整的答案,并详细介绍Airflow如何为新手工作.

当然,你首先需要其中一个:

from airflow.operators.python_operator import PythonOperator

class SQLTemplatedPythonOperator(PythonOperator):

    # somehow ('.sql',) doesn't work but tuple of two works...
    template_ext = ('.sql','.abcdefg')

假设您有一个如下所示的sql模板文件:

# stored at path: $AIRFLOW_HOME/sql/some.sql
select {{some_params}} from my_table;

首先确保将文件夹添加到dag参数的搜索路径中.

不要将template_searchpath传递给args,然后将args传递给DAG!它不起作用.

dag = DAG(
    dag_id= "some_name",
    default_args=args,
    schedule_interval="@once",
    template_searchpath='/Users/your_name/some_path/airflow_home/sql'
)

那么您的操作员电话将是

SQLTemplatedPythonOperator(
        templates_dict={'query': 'some.sql'},
        op_kwargs={"args_directly_passed_to_your_function": "some_value"},
        task_id='dummy',
        params={"some_params":"some_value"},
        python_callable=your_func,
        provide_context=True,
        dag=dag,
    )

你的功能将是:

def your_func(args_directly_passed_to_your_function=None):
    query = context['templates_dict']['query']
    dome_some_thing(query)

一些解释:

Airflow使用上下文中的值来呈现模板.要手动将其添加到上下文中,您可以使用上面的params字段.

PythonOperator不再从template_ext字段获取模板文件扩展名,就像@Ardan提到的那样.源代码在这里.它只需要从self .__ class __.template_ext扩展.

气流循环遍历template_dict字段,如果value.endswith(file_extension)== True,则呈现模板.

我认为这不可能.但以下解决方法可能会有所帮助:

def templated_function(ds, **kwargs):
    kwargs['ds'] = ds                                # put ds into 'context'
    task = kwargs['task']                            # get handle on task
    templ = open(kwargs['templates_dict']['file1']).read() # get template
    sql = task.render_template('', tmpl, kwargs)           # render it
    pp.pprint(sql)

但是,我会喜欢更好的解决方案!