气流中的辅助功能

Question

辅助函数应该放在气流中的什么位置？是否应该添加到插件目录中？我将为我的 ETL 作业编写一些可重复的代码，如下所示：

#create a sub dag containing DROP -> CREATE -> INSERT 
sub_create_dag = DAG('%s.%s' % (parent_dag_name, child_dag_name), default_args=args)
pg_drop = DropPostgresOperator(task_id='drop_{}'.format(table), sql="", params={'schema': schema, 'table': table}, postgres_conn_id=args['connection_id'], autocommit=True, dag=sub_create_dag)
pg_create = PostgresOperator(task_id='create_{}'.format(table), sql='{sql_path}/create_{name}.sql'.format(sql_path=sql_path, name=table), postgres_conn_id=args['connection_id'], autocommit=True, dag=sub_create_dag)
pg_insert = PostgresOperator(task_id='insert_{}'.format(table), sql='{sql_path}/insert_{name}.sql'.format(sql_path=sql_path, name=table), postgres_conn_id=args['connection_id'], autocommit=True, dag=sub_create_dag)
pg_drop >> pg_create >> pg_insert
return dag

问题是我会遇到如下错误：

File "/usr/local/lib/python2.7/dist-packages/jinja2/loaders.py", line 187, in get_source   
raise TemplateNotFound(template)

Answer 1

我们结合使用了两者。我们在插件文件夹中将独立任务实现为自定义运算符，而在 DAG 文件夹本身中将小函数实现为 python 函数。关于错误，这是因为jinja找不到templates文件夹。如果您在 DAG 或插件文件夹中实现辅助函数，则不应弹出此错误。