python - 如何在 Airflow 中基于另一个 AWS-glue 任务成功完成在 Airflow 中启动 python operator boto3 AWS-glue 任务？

Question

我的 Ariflow 脚本有几个任务使用 python 运算符来启动使用 boto3 的粘合作业。尽管流程取决于 Ariflow 中的每个任务。后续任务不会等到前一个任务成功完成。

一旦调用前一个粘合作业任务，就会调用下一个粘合作业任务。最后，即使 Airflow 看起来已成功完成，粘合作业仍在运行几分钟。

    from airflow import DAG
    from airflow.operators.python_operator import PythonOperator
    from datetime import datetime, timedelta
    from airflow.operators.sensors import TimeDeltaSensor
    import boto3
    import json

    default_args = {
    'owner': 'SAM',
    'start_date': datetime(2019, 5, 27)
     }


    glue_client = boto3.client('glue', region_name='us-east-1')

    def run_customer_job():
        glue_client.start_job_run(JobName='customer')  


    def run_product_job():
        glue_client.start_job_run(JobName='product')



    with DAG('dataload', default_args=default_args, schedule_interval="0 15 * * *") as dag:
        task1 = PythonOperator(task_id='task1',                                              
                               python_callable=run_customer_job)
        task2 = PythonOperator(task_id='task2',                                                          
                               python_callable=run_product_job)



    task1 >> task2 

Answer 1

创建一个新的 GlueOperator，基于 https://github.com/apache/airflow/blob/master/airflow/contrib/operators/awsbatch_operator.py