Azure 数据工厂和 Azure Synapse Analytics 中的管道和活动

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

本文帮助你了解 Azure 数据工厂和 Azure Synapse Analytics 中的管道和活动，并帮助你利用它们为数据移动和数据处理方案构造端到端数据驱动工作流。

概述

一个数据工厂或 Synapse 工作区可以有一个或多个管道。 “管道”是共同执行一项任务的活动的逻辑分组。 例如，管道可以包含一组活动，这些活动引入和清除日志数据，然后启动映射数据流以分析日志数据。 可以通过管道将活动作为一个集来管理，而非单独管理每个活动。 可以部署和计划管道，而不需单独对活动进行操作。

管道中的活动定义对数据执行的操作。 例如，可使用复制活动将数据从 SQL Server 复制到 Azure Blob 存储。 然后，使用数据流活动或 Databricks Notebook 活动来处理数据并将数据从 Blob 存储转换为 Azure Synapse Analytics 池，在此池的基础上构建商业智能报表解决方案。

Azure 数据工厂和 Azure Synapse Analytics 支持三组活动：数据移动活动、数据转换活动和控制活动。 每个活动可获取零个或多个输入数据集，并生成一个或多个输出数据集。 下图显示了管道、活动和数据集之间的关系：

输入数据集表示管道中活动的输入，输出数据集表示活动的输出。 数据集可识别不同数据存储（如表、文件、文件夹和文档）中的数据。 创建数据集后，可将其与管道中的活动一起使用。 例如，数据集可以是复制活动或 HDInsightHive 活动的输入/输出数据集。 有关数据集的详细信息，请参阅 Azure 数据工厂中的数据集一文。

数据移动活动

数据工厂中的复制活动可以将数据从源数据存储复制到接收器数据存储。 数据工厂支持本部分中的表中列出的数据存储。 来自任何源的数据都可以写入到任何接收器。

有关详细信息，请参阅复制活动 - 概述一文。

单击某个数据存储即可了解如何将数据复制到该存储，以及如何从该存储复制数据。

数据转换活动

Azure 数据工厂和 Azure Synapse Analytics 支持以下转换活动，这些活动既可以单独添加，也可以与其他活动关联在一起添加。

有关详细信息，请参阅数据转换活动一文。

控制流活动

支持以下控制流活动：

使用 UI 创建管道

管道 JSON

下面介绍如何以 JSON 格式定义管道：

{
    "name": "PipelineName",
    "properties":
    {
        "description": "pipeline description",
        "activities":
        [
        ],
        "parameters": {
        },
        "concurrency": <your max pipeline concurrency>,
        "annotations": [
        ]
    }
}

活动 JSON

activities 节中可定义有一个或多个活动。 有两种主要类型的活动：执行和控制活动。

执行活动

执行活动包括数据移动和数据转换活动。 它们具有以下顶级结构：

{
    "name": "Execution Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "linkedServiceName": "MyLinkedService",
    "policy":
    {
    },
    "dependsOn":
    {
    }
}

下表描述了活动 JSON 定义中的属性：

活动策略

影响活动运行时行为的策略，提供配置性选项。 活动策略仅对执行活动可用。

活动策略 JSON 定义

{
    "name": "MyPipelineName",
    "properties": {
      "activities": [
        {
          "name": "MyCopyBlobtoSqlActivity",
          "type": "Copy",
          "typeProperties": {
            ...
          },
         "policy": {
            "timeout": "00:10:00",
            "retry": 1,
            "retryIntervalInSeconds": 60,
            "secureOutput": true
         }
        }
      ],
        "parameters": {
           ...
        }
    }
}

控制活动

控制活动具有以下顶级结构：

{
    "name": "Control Activity Name",
    "description": "description",
    "type": "<ActivityType>",
    "typeProperties":
    {
    },
    "dependsOn":
    {
    }
}

活动依赖项

活动依赖项定义后续活动对以前活动的依赖方式，确定是否继续执行下一个任务的条件。 活动可能依赖于具有不同依赖项条件的一个或多个以前的活动。

不同依赖项条件有：成功、失败、跳过、完成。

例如，如果管道具有活动 A -> 活动 B，则可能发生的不同情况是：

• 活动 B 对活动 A 具有依赖项条件“成功”：只有活动 A 的最终状态为“成功”，活动 B 才运行
• 活动 B 对活动 A 具有依赖项条件“失败”：只有活动 A 的最终状态为“失败”，活动 B 才运行
• 活动 B 对活动 A 具有依赖项条件“完成”：如果活动 A 的最终状态为“成功”或“失败”，则活动 B 运行
• 活动 B 对活动 A 具有依赖项条件“跳过”：如果活动 A 的最终状态为“跳过”，则活动 B 运行。 在活动 X -> 活动 Y -> 活动 Z 的情况下出现跳过，其中每个活动仅在以前的活动成功后才运行。 如果活动 X 失败，则活动 Y 的状态为“跳过”，因为它从不执行。 类似地，活动 Z 的状态也为“跳过”。

示例：活动 2 是否运行取决于活动 1 是否成功运行

{
    "name": "PipelineName",
    "properties":
    {
        "description": "pipeline description",
        "activities": [
         {
            "name": "MyFirstActivity",
            "type": "Copy",
            "typeProperties": {
            },
            "linkedServiceName": {
            }
        },
        {
            "name": "MySecondActivity",
            "type": "Copy",
            "typeProperties": {
            },
            "linkedServiceName": {
            },
            "dependsOn": [
            {
                "activity": "MyFirstActivity",
                "dependencyConditions": [
                    "Succeeded"
                ]
            }
          ]
        }
      ],
      "parameters": {
       }
    }
}

复制管道示例

在以下示例管道中，activities 节有一个 Copy 类型的活动。 在此示例中，复制活动将 Azure Blob 存储中的数据复制到 Azure SQL 数据库中的数据库。

{
  "name": "CopyPipeline",
  "properties": {
    "description": "Copy data from a blob to Azure SQL table",
    "activities": [
      {
        "name": "CopyFromBlobToSQL",
        "type": "Copy",
        "inputs": [
          {
            "name": "InputDataset"
          }
        ],
        "outputs": [
          {
            "name": "OutputDataset"
          }
        ],
        "typeProperties": {
          "source": {
            "type": "BlobSource"
          },
          "sink": {
            "type": "SqlSink",
            "writeBatchSize": 10000,
            "writeBatchTimeout": "60:00:00"
          }
        },
        "policy": {
          "retry": 2,
          "timeout": "01:00:00"
        }
      }
    ]
  }
}

请注意以下几点：

• 在 activities 节中，只有一个活动的 type 设置为 Copy。
• 活动的输入设置为 InputDataset，活动的输出设置为 OutputDataset。 有关在 JSON 中定义数据集的信息，请参阅数据集文章。
• 在 typeProperties 节中，BlobSource 指定为源类型，SqlSink 指定为接收器类型。 在数据移动活动部分，单击想要用作源或接收器的数据存储，以了解有关将数据移动到该数据存储或从该数据存储移出的详细信息。

有关创建此管道的完整演练，请参阅快速入门：创建数据工厂。

转换管道示例

在以下示例管道中，activities 节有一个 HDInsightHive 类型的活动。 在此示例中，HDInsight Hive 活动通过在 Azure HDInsight Hadoop 群集上运行 Hive 脚本文件，转换 Azure Blob 存储中的数据。

{
    "name": "TransformPipeline",
    "properties": {
        "description": "My first Azure Data Factory pipeline",
        "activities": [
            {
                "type": "HDInsightHive",
                "typeProperties": {
                    "scriptPath": "adfgetstarted/script/partitionweblogs.hql",
                    "scriptLinkedService": "AzureStorageLinkedService",
                    "defines": {
                        "inputtable": "wasb://adfgetstarted@<storageaccountname>.blob.core.chinacloudapi.cn/inputdata",
                        "partitionedtable": "wasb://adfgetstarted@<storageaccountname>.blob.core.chinacloudapi.cn/partitioneddata"
                    }
                },
                "inputs": [
                    {
                        "name": "AzureBlobInput"
                    }
                ],
                "outputs": [
                    {
                        "name": "AzureBlobOutput"
                    }
                ],
                "policy": {
                    "retry": 3
                },
                "name": "RunSampleHiveActivity",
                "linkedServiceName": "HDInsightOnDemandLinkedService"
            }
        ]
    }
}

请注意以下几点：

• 在 activities 节中，只有一个活动的 type 设置为 HDInsightHive。
• Hive 脚本文件 partitionweblogs.hql 存储在 Azure 存储帐户（由 scriptLinkedService 指定，名为 AzureStorageLinkedService）中，以及 adfgetstarted 容器的 script 文件夹中。
• defines 部分用于指定以配置单元配置值传递到配置单元脚本的运行时设置（例如，${hiveconf:inputtable}，${hiveconf:partitionedtable}）。

每个转换活动的 typeProperties 节都不同。 若要了解有关转换活动所支持的类型属性的详细信息，请单击数据转换活动中的转换活动。

有关创建此管道的完整演练，请参阅教程：使用 Spark 转换数据。

管道中的多个活动

前两个示例管道中只有一个活动。 可在管道中添加多个活动。 如果在管道中具有多个活动且后续活动不依赖于以前的活动，则活动可能并行运行。

可以使用活动依赖项将两个活动进行链接，以定义后续活动对以前活动的依赖方式，确定是否继续执行下一个任务的条件。 活动可以依赖于具有不同依赖项条件的一个或多个以前的活动。

计划管道

管道由触发器计划 触发器有多种类型（让管道按时钟计划触发的计划触发器，以及按需触发管道的手动触发器）。 有关触发器的详细信息，请参阅管道执行和触发器一文。

若要使你的触发器启动管道运行，必须包含对触发器定义中的特定管道的管道引用。 管道和触发器具有 n-m 关系。 多个触发器可以启动单个管道，同一个触发器可以启动多个管道。 定义管道后，必须启动触发器，以使其开始触发管道。 有关触发器的详细信息，请参阅管道执行和触发器一文。

例如，假设有一个计划触发器“触发器 A”，我希望使用该触发器启动我的管道“MyCopyPipeline”。 定义该触发器，如以下示例中所示：

触发器 A 定义

{
  "name": "TriggerA",
  "properties": {
    "type": "ScheduleTrigger",
    "typeProperties": {
      ...
      }
    },
    "pipeline": {
      "pipelineReference": {
        "type": "PipelineReference",
        "referenceName": "MyCopyPipeline"
      },
      "parameters": {
        "copySourceName": "FileSource"
      }
    }
  }
}

相关内容

• 构建包含复制活动的管道
• 使用数据转换活动生成管道