如何根据另一个AWS-Glue任务在气流中成功完成，如何在气流中启动Python操作员Boto3 AWS-Glue任务

Question

我的Ariflow脚本使用Python操作员使用几个任务，使用BOTO3启动胶合作业。即使流程取决于Ariflow中的每个任务。连续任务不会等到上一个任务成功完成。

调用上一个胶水工作任务后，下一个胶水工作任务就会立即调用。最后，即使气流看起来成功完成了胶水工作仍在运行几分钟。

    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

基于https://github.com/apache/apache/airflow/main/main/main/main/main/airflow/mazon/amazon/amazon/aws/aperators/glue.py.py.py.py.pypy<<p>创建一个新的Glueoperator>

Answer 2

我将在调用start_job_run之后创建一个状态循环，这样您的任务才能完成。

快速简单的示例(您需要处理失败的任务等(：

job = glue_client.start_job_run(JobName='customer')
while True:
  status = glue.get_job_run(JobName=job['Name'], RunId=job['JobRunId'])
  if status['JobRun']['JobRunState'] == 'SUCCEEDED':
    break
  time.sleep(10)