我已经配置了airflow并创建了一些DAG和子DAG,这些DAG和子DAG调用了几个操作符。
我的问题是,当一个操作符运行并完成任务时,我希望以某种python结构接收结果。 例如:
文件1.py
...
...
sub_dag_one=SubDagOperator(subdag=subdag_accessHive(
PARENT_DAG_NAME, CHILD_DAG_NAME, default_args, STEP, macros,path,
),
task_id=DELP_DAG_NAME,
dag=dag,
)
文件2.py
from airflow import DAG
from airflow.operators import HiveOperator
def subdag_callHive(parent, child, args, step,
user_defined_macros, path
):
dag_subdag = DAG(
dag_id='%s.%s' % (parent, child),
default_args=args,
schedule_interval="@daily",
template_searchpath=path,
user_defined_macros=user_defined_macros,
)
# some work...
HiveOperator(
task_id='some_id',
hiveconf_jinja_translate=True,
hql='select field1 from public.mytable limit 4;',
trigger_rule='all_done',
dag=dag_subdag,
)
return dag_subdag
函数subdag_callHive是从另一个python脚本调用的,在该脚本中定义了主Dag,并需要所有其他参数。
我只想从HiveOperator那里得到结果 (*select*from public.mytable limit 4;*)在本例中为4个值。
返回的dag_subdag是一个对象<;class'airlow.models.dag'>;,包含传递给调用的所有属性/数据,但不包含HiveOperator所做操作的信息。
这可能吗?如果是的话,怎么能做到。
你可以根据需要用钩子。基本上HiveOperator也这样做,他调用Hive钩子,钩子有多个方法来处理结果。
使用PythonOperator调用一个函数,然后启动一个hive钩子。
下面的示例可能会帮助您。
代码段:
所有可用的方法都可以在这里找到:https://github.com/apache/incubator-airflow/blob/master/airflow/hooks/hive_hooks.py
相关问题 更多 >
编程相关推荐