Was ist die Art, Parameter in abhängige Aufgaben in Airflow übergeben? Ich habe viele Bashes-Dateien, und ich versuche, diesen Ansatz auf den Luftstrom zu migrieren, aber ich weiß nicht, wie ich einige Eigenschaften zwischen Aufgaben weitergeben soll.Luftstrom Pass Parameter zu abhängigen Aufgabe
Dies ist ein echtes Beispiel:
#sqoop bash template
sqoop_template = """
sqoop job --exec {{params.job}} -- --target-dir {{params.dir}} --outdir /src/
"""
s3_template = """
s3-dist-cp --src= {{params.dir}} "--dest={{params.s3}}
"""
#Task of extraction in EMR
t1 = BashOperator(
task_id='extract_account',
bash_command=sqoop_template,
params={'job': 'job', 'dir': 'hdfs:///account/' + time.now().strftime("%Y-%m-%d-%H-%M-%S")},
dag=dag)
#Task to upload in s3 backup.
t2 = BashOperator(
task_id='s3_upload',
bash_command=s3_template,
params={}, #here i need the dir name created in t1
depends_on_past=True
)
t2.set_upstream(t1)
In t2 Ich brauche die Verzeichnisnamen in t1 erstellt zuzugreifen.
Lösung
#Execute a valid job sqoop
def sqoop_import(table_name, job_name):
s3, hdfs = dirpath(table_name)
sqoop_job = job_default_config(job_name, hdfs)
#call(sqoop_job)
return {'hdfs_dir': hdfs, 's3_dir': s3}
def s3_upload(**context):
hdfs = context['task_instance'].xcom_pull(task_ids='sqoop_import')['hdfs_dir']
s3 = context['task_instance'].xcom_pull(task_ids='sqoop_import')['s3_dir']
s3_cpdist_job = ["s3-dist-cp", "--src=%s" % (hdfs), "--dest=%s" % (s3)]
#call(s3_cpdist_job)
return {'s3_dir': s3} #context['task_instance'].xcom_pull(task_ids='sqoop_import')
def sns_notify(**context):
s3 = context['task_instance'].xcom_pull(task_ids='distcp_s3')['s3_dir']
client = boto3.client('sns')
arn = 'arn:aws:sns:us-east-1:744617668409:pipeline-notification-stg'
response = client.publish(TargetArn=arn, Message=s3)
return response
Das ist die endgültige Lösung nicht der Fall ist, so Verbesserungen sind willkommen. Vielen Dank.
Eine Lösung, meiner Meinung nach, ist es, eine Datei mit den in t1 erstellten Eigenschaften zu erstellen und diese Datei in t2 zu konsumieren. –