使用工作流编排管理器运行现有管道

适用于：Azure 数据工厂 Azure Synapse Analytics

数据工厂管道提供 100 多个数据源连接器，这些连接器提供可缩放且可靠的数据集成/数据流。 在某些情况下，你想要从 Apache Airflow DAG 运行现有的数据工厂管道。 本教程将向您展示如何操作。

先决条件

• Azure 订阅。 如果还没有 Azure 订阅，可以在开始前创建一个免费 Azure 帐户。
• Azure 存储帐户。 如果没有存储帐户，请参阅创建 Azure 存储帐户以获取创建步骤。 确保存储帐户仅允许来自选定的网络的访问。
• Azure 数据工厂管道。 可以按照任何这些教程新建一个数据工厂管道（如果还没有），或者在开始试用第一个数据工厂管道中一键选择创建一个管道。
• 设置服务主体。 需要创建一个新的服务主体或使用现有服务主体，并授予其运行管道的权限（例如，现有管道所在的数据工厂中的参与者角色），即使工作流编排管理器环境和管道存在于同一数据工厂中也是如此。 需要获取服务主体的客户端 ID 和客户端密码（API 密钥）。

步骤

1. 使用以下内容新建一个 Python 文件 adf.py：

   from datetime import datetime, timedelta
   
   from airflow.models import DAG, BaseOperator
   
   try:
       from airflow.operators.empty import EmptyOperator
   except ModuleNotFoundError:
       from airflow.operators.dummy import DummyOperator as EmptyOperator  # type: ignore
   from airflow.providers.microsoft.azure.operators.data_factory import AzureDataFactoryRunPipelineOperator
   from airflow.providers.microsoft.azure.sensors.data_factory import AzureDataFactoryPipelineRunStatusSensor
   from airflow.utils.edgemodifier import Label
   
   with DAG(
       dag_id="example_adf_run_pipeline",
       start_date=datetime(2022, 5, 14),
       schedule_interval="@daily",
       catchup=False,
       default_args={
           "retries": 1,
           "retry_delay": timedelta(minutes=3),
           "azure_data_factory_conn_id": "<connection_id>", #This is a connection created on Airflow UI
           "factory_name": "<FactoryName>",  # This can also be specified in the ADF connection.
           "resource_group_name": "<ResourceGroupName>",  # This can also be specified in the ADF connection.
       },
       default_view="graph",
   ) as dag:
       begin = EmptyOperator(task_id="begin")
       end = EmptyOperator(task_id="end")
   
       # [START howto_operator_adf_run_pipeline]
       run_pipeline1: BaseOperator = AzureDataFactoryRunPipelineOperator(
           task_id="run_pipeline1",
           pipeline_name="<PipelineName>",
           parameters={"myParam": "value"},
       )
       # [END howto_operator_adf_run_pipeline]
   
       # [START howto_operator_adf_run_pipeline_async]
       run_pipeline2: BaseOperator = AzureDataFactoryRunPipelineOperator(
           task_id="run_pipeline2",
           pipeline_name="<PipelineName>",
           wait_for_termination=False,
       )
   
       pipeline_run_sensor: BaseOperator = AzureDataFactoryPipelineRunStatusSensor(
           task_id="pipeline_run_sensor",
           run_id=run_pipeline2.output["run_id"],
       )
       # [END howto_operator_adf_run_pipeline_async]
   
       begin >> Label("No async wait") >> run_pipeline1
       begin >> Label("Do async wait with sensor") >> run_pipeline2
       [run_pipeline1, pipeline_run_sensor] >> end
   
       # Task dependency created via `XComArgs`:
       #   run_pipeline2 >> pipeline_run_sensor
   

   必须使用工作流编排管理器 UI“管理员”->“连接”->“+”-> 选择“连接类型”作为“Azure 数据工厂”创建连接，然后填写 client_id、client_secret、tenant_id、subscription_id、resource_group_name、data_factory_name 和 pipeline_name。

2. 将 adf.py 文件上传到名为 DAGS 的文件夹中的 blob 存储。

3. 将 DAGS 文件夹导入工作流编排管理器环境。 如果没有，请新建一个

   

相关内容

• 工作流编排管理器定价
• 更改工作流编排管理器环境的密码