azure - 集成 Azure 数据工厂和 Airflow

Question

我的最终目标是使用 Airflow 运行 Azure 数据工厂 (ADF) 管道。我当前的设置是一个 docker 文件，其中包含所需的 python 包，例如 azure 数据提供程序和来自 apache airflow 的 helm 图表。我有一个带有 celery 执行器的自定义 value.yaml，每次我都会升级它以在本地运行 Airflow 。

到目前为止这部分是成功的。

Python 包提供了一个 azure 的数据工厂选项，就像我想要的那样，但我无法验证这些选项。我使用 python 代码测试了相同的凭据(资源组、客户端 ID、客户端 key 、租户 ID、订阅 ID)，它是有效的。

这在 Airflow 上不起作用，我看到此错误

*** Trying to get logs (last 100 lines) from worker pod azuredatafactoryrunpipeline.3f394072d09e4c6e8d59566776f18b78 ***

*** Unable to fetch logs from worker pod azuredatafactoryrunpipeline.3f394072d09e4c6e8d59566776f18b78 ***
(404)
Reason: Not Found

HTTP response headers: HTTPHeaderDict({'Audit-Id': '4a506d8b-9bac-44d8-8a7c-ee01810e478f', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'Date': 'Mon, 07 Feb 2022 22:05:52 GMT', 'Content-Length': '288'})

HTTP response body: b'{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"pods \\"azuredatafactoryrunpipeline.3f394072d09e4c6e8d59566776f18b78\\" not found","reason":"NotFound","details":{"name":"azuredatafactoryrunpipeline.3f394072d09e4c6e8d59566776f18b78","kind":"pods"},"code":404}\n'

如何调试这个？

是否可以通过 Airflow 测试 ADF 连接？

Answer 1

Airflow 有 AzureDataFactoryRunPipelineOperator，用于在数据工厂内执行管道(请参阅 docs)

您还可以添加传感器，等待管道完成后再继续执行 Airflow 工作流程。

使用示例:

from airflow.providers.microsoft.azure.operators.data_factory import AzureDataFactoryRunPipelineOperator
from airflow.providers.microsoft.azure.sensors.data_factory import AzureDataFactoryPipelineRunStatusSensor

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

pipeline_run_sensor = AzureDataFactoryPipelineRunStatusSensor(
    task_id="pipeline_run_sensor",
    run_id=run_pipeline.output["run_id"],
)
run_pipeline2 >> pipeline_run_sensor

要在 Airflow 中定义 ADF 连接，请检查此 doc .