W systemie Airflow napotykam problem, który muszę przekazać job_flow_id
do jednego z moich kroków emr. Jestem w stanie pobrać job_flow_id
od operatora, ale kiedy mam zamiar utworzyć kroki do przesłania do klastra, wartość task_instance
jest nieprawidłowa. Mam następujący kod:Przepływ powietrza - Instancja zadania w sterowniku EMR
def issue_step(name, args):
return [
{
"Name": name,
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "s3://....",
"Args": args
}
}
]
dag = DAG('example',
description='My dag',
schedule_interval='0 8 * * 6',
dagrun_timeout=timedelta(days=2))
try:
create_emr = EmrCreateJobFlowOperator(
task_id='create_job_flow',
aws_conn_id='aws_default',
dag=dag
)
load_data_steps = issue_step('load', ['arg1', 'arg2'])
load_data_steps[0]["HadoopJarStep"]["Args"].append('--cluster-id')
load_data_steps[0]["HadoopJarStep"]["Args"].append(
"{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}") # the value here is not exchanged with the actual job_flow_id
load_data = EmrAddStepsOperator(
task_id='load_data',
job_flow_id="{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}", # this is correctly exchanged with the job_flow_id - same for the others
aws_conn_id='aws_default',
steps=load_data_steps,
dag=dag
)
check_load_data = EmrStepSensor(
task_id='watch_load_data',
job_flow_id="{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}",
step_id="{{ task_instance.xcom_pull('load_data', key='return_value')[0] }}",
aws_conn_id='aws_default',
dag=dag
)
cluster_remover = EmrTerminateJobFlowOperator(
task_id='remove_cluster',
job_flow_id="{{ task_instance.xcom_pull('create_job_flow', key='return_value') }}",
aws_conn_id='aws_default',
dag=dag
)
create_emr_recommendations >> load_data
load_data >> check_load_data
check_load_data >> cluster_remover
except AirflowException as ae:
print ae.message
Problem polega na tym, że kiedy sprawdzić EMR, zamiast widząc --cluster-id j-1234
w kroku load_data
widzę --cluster-id "{{task_instance.xcom_pull('create_job_flow', key='return_value')}}"
, co powoduje mój krok na niepowodzenie.
Jak mogę uzyskać rzeczywistą wartość wewnątrz funkcji krokowej?
Dzięki i wesołych świąt
czy próbowałeś dodać wartość bez cudzysłowów? load_data_steps [0] ["HadoopJarStep"] ["Args"]. append ( {{task_instance.xcom_pull ('create_job_flow', key = 'return_value')}}) –
gdzie otrzymam '' 'task_instance''' obiekt z? Nadal uczę się go używać. – davideberdin