Azure Data Factory 操作符

Azure Data Factory 是 Azure 的云 ETL 服务，用于大规模无服务器数据集成和数据转换。它提供了一个无代码 UI，用于直观创作和单一窗格监控和管理。

AzureDataFactoryRunPipelineOperator

使用 AzureDataFactoryRunPipelineOperator 在数据工厂中执行管道。默认情况下，操作符将定期检查已执行管道的状态，以“成功”状态终止。此功能可以禁用以进行异步等待（通常使用 AzureDataFactoryPipelineRunStatusSensor），方法是将 wait_for_termination 设置为 False。

以下是使用此操作符执行 Azure Data Factory 管道的示例。

tests/system/providers/microsoft/azure/example_adf_run_pipeline.py[源代码]

    run_pipeline1 = AzureDataFactoryRunPipelineOperator(
        task_id="run_pipeline1",
        pipeline_name="pipeline1",
        parameters={"myParam": "value"},
    )

以下是使用此操作符执行 Azure Data Factory 管道的示例，其中包含可延期标志，以便在 Airflow 触发器上轮询管道运行的状态。

tests/system/providers/microsoft/azure/example_adf_run_pipeline.py[源代码]

run_pipeline3 = AzureDataFactoryRunPipelineOperator(
    task_id="run_pipeline3",
    pipeline_name="pipeline1",
    parameters={"myParam": "value"},
    deferrable=True,
)

以下是使用此运算符执行管道但与 AzureDataFactoryPipelineRunStatusSensor 结合使用以执行异步等待的不同示例。

tests/system/providers/microsoft/azure/example_adf_run_pipeline.py[源代码]

    run_pipeline2 = AzureDataFactoryRunPipelineOperator(
        task_id="run_pipeline2",
        pipeline_name="pipeline2",
        wait_for_termination=False,
    )

    pipeline_run_sensor = AzureDataFactoryPipelineRunStatusSensor(
        task_id="pipeline_run_sensor",
        run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
    )

    # Performs polling on the Airflow Triggerer thus freeing up resources on Airflow Worker
    pipeline_run_sensor_deferred = AzureDataFactoryPipelineRunStatusSensor(
        task_id="pipeline_run_sensor_defered",
        run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
        deferrable=True,
    )

    pipeline_run_async_sensor = AzureDataFactoryPipelineRunStatusSensor(
        task_id="pipeline_run_async_sensor",
        run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
        deferrable=True,
    )

如果希望在传感器运行时释放工作程序槽，还可以在 AzureDataFactoryPipelineRunStatusSensor 中使用可延迟模式。

tests/system/providers/microsoft/azure/example_adf_run_pipeline.py[源代码]

    run_pipeline2 = AzureDataFactoryRunPipelineOperator(
        task_id="run_pipeline2",
        pipeline_name="pipeline2",
        wait_for_termination=False,
    )

    pipeline_run_sensor = AzureDataFactoryPipelineRunStatusSensor(
        task_id="pipeline_run_sensor",
        run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
    )

    # Performs polling on the Airflow Triggerer thus freeing up resources on Airflow Worker
    pipeline_run_sensor_deferred = AzureDataFactoryPipelineRunStatusSensor(
        task_id="pipeline_run_sensor_defered",
        run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
        deferrable=True,
    )

    pipeline_run_async_sensor = AzureDataFactoryPipelineRunStatusSensor(
        task_id="pipeline_run_async_sensor",
        run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
        deferrable=True,
    )

异步轮询数据工厂管道运行状态

使用 AzureDataFactoryPipelineRunStatusAsyncSensor（可延迟版本）以定期异步检索数据工厂管道运行的状态。此传感器将释放工作程序槽，因为在 Airflow 触发器上轮询作业状态，从而有效利用 Airflow 中的资源。

tests/system/providers/microsoft/azure/example_adf_run_pipeline.py[源代码]

run_pipeline2 = AzureDataFactoryRunPipelineOperator(
    task_id="run_pipeline2",
    pipeline_name="pipeline2",
    wait_for_termination=False,
)

pipeline_run_sensor = AzureDataFactoryPipelineRunStatusSensor(
    task_id="pipeline_run_sensor",
    run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
)

# Performs polling on the Airflow Triggerer thus freeing up resources on Airflow Worker
pipeline_run_sensor_deferred = AzureDataFactoryPipelineRunStatusSensor(
    task_id="pipeline_run_sensor_defered",
    run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
    deferrable=True,
)

pipeline_run_async_sensor = AzureDataFactoryPipelineRunStatusSensor(
    task_id="pipeline_run_async_sensor",
    run_id=cast(str, XComArg(run_pipeline2, key="run_id")),
    deferrable=True,
)

参考

有关详细信息，请参阅 Microsoft 文档

此条目是否有用？