4

气流中的辅助功能应该放在哪里？它应该添加到插件目录中吗？我将为我的 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)
4

1 回答 1

4

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

于 2017-06-26T08:58:44.493 回答