如何设置气流DAG以在触发下一个数据流作业之前完成上一个数据流任务？

Question

我在Google Cloud Composer中从Airflow调用数据流作业，

a >> b >> c

a、 b和c是调用数据流作业的任务。我只想在数据流作业完成后运行b，问题是它们都同时运行。

我怎么能等到之前的工作完成？

Answer 1

放置作业后，您需要放置一个传感器，以验证作业是否完成。

例子：

   start_python_job_async = DataflowCreatePythonJobOperator(
        task_id="start-python-job-async",
        py_file=GCS_PYTHON,
        py_options=[],
        job_name='{{task.task_id}}',
        options={
            'output': GCS_OUTPUT,
        },
        py_requirements=['apache-beam[gcp]==2.25.0'],
        py_interpreter='python3',
        py_system_site_packages=False,
        location='europe-west3',
        wait_until_finished=False,
    )

    wait_for_python_job_async_done = DataflowJobStatusSensor(
        task_id="wait-for-python-job-async-done",
        job_id="{{task_instance.xcom_pull('start-python-job-async')['job_id']}}",
        expected_statuses={DataflowJobStatus.JOB_STATE_DONE},
        location='europe-west3',
    )

    start_python_job_async >> wait_for_python_job_async_done

您可以查看进一步解释的文档和示例