我正在尝试使用 Airflow 在 AWS EMR 上运行 11 步,并遵循此代码作为参考。由于使用 EmrAddStepsOperator 和 EmrStepSensor 进行 11 步会重复太多。所以我试图遍历它。我在我的 DAG 中使用了以下代码。
step_adder = list()
step_checker = list()
steps = ['step1', 'step2', 'step3', 'step4', 'step5', 'step6'...till step11]
# @evalcontextfilter
# def dangerous_render(context, value):
# return Markup(Template(value).render(context)).render()
for i in range(0,len(steps)):
#Add step
step_adder.append(EmrAddStepsOperator(
task_id=steps[i],
job_flow_id="{{ task_instance.xcom_pull(task_ids='create_job_flow', key='return_value') }}",
aws_conn_id='aws_default',
steps=eval('step_'+str(i+1)),
))
print(step_adder)
#Step Sensor for checking
step_checker.append(EmrStepSensor(
task_id=steps[i]+'_check',
job_flow_id="{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}",
#step_id="{{"task_instance.xcom_pull(task_ids={}, key='return_value')[0]",steps[i]}}",
step_id='(Template("{{ "task_instance.xcom_pull(task_ids=params.step, key='return_value')[0] }}").render({'params': {'step': steps[i]}}))',
aws_conn_id='aws_default',
))
我在这里遇到一个错误,EmrStepSensor 期望来自 EMR 的 step_id 在这里输入,并且是从 xcom 获取的生成的(我猜,我不是 100% 确定这段代码是如何工作的)。但是我的步骤存储在步骤列表中,因此我无法在 step_id 的 task_id 中给出静态值,就像在参考代码中给出的那样,我无法弄清楚如何使用带有 python 变量值的 jinja 模板将值放在这里从步骤列表中。
我使用了以下两种方式,以便 step_id 可以根据 steps[i] 中的步骤名称从 EMR 中获取正确的步骤
step_id="{{"task_instance.xcom_pull(task_ids={}, key='return_value')[0]",steps[i]}}",
step_id='(Template("{{ "task_instance.xcom_pull(task_ids=params.step, key='return_value')[0] }}")
然而,这两个都失败了,在 Airflow 中出现语法错误。因此,如果有人能指出我这样做的正确方向,我将不胜感激。我正在使用 Airflow 1.10.12(这是 AWS 上托管 Apache Airflow 中 Airflow 的默认版本)。