数据集成工具：Azure Data Factory：5.使用映射数据流进行数据转换.docx

数据集成工具：Azure Data Factory：5.使用映射数据流进行

数据转换

1 理解映射数据流

1.1 映射数据流的概念

映射数据流是 Azure Data Factory 中用于数据转换的一种强大工具。它允许

用户通过图形界面设计数据转换逻辑，而无需编写复杂的代码。映射数据流支

持多种数据转换操作，如选择、过滤、聚合、连接等，可以处理结构化和半结

映射数据流与 Azure Data Factory 中的另一种数据转换工具——数据流（也

称为“数据流活动”），在概念上是相同的，但它们在处理数据的方式上有所不

同：

� 映射数据流：使用基于列的转换，这意味着数据流中的每个转换

操作都是针对数据集中的列进行的。映射数据流适用于处理结构化数据，

如关系型数据库中的数据，或 CSV、JSON 等格式的文件数据。它支持丰

富的数据转换操作，如选择、过滤、聚合、连接等。

� 数据流（非映射）：使用基于行的转换，这意味着数据流中的每个

转换操作都是针对数据集中的行进行的。非映射数据流适用于处理非结

构化或半结构化数据，如文本文件、日志文件等。它支持的数据转换操

3, Charlie, 35, Chicago

我们想要从这个 CSV 文件中选择 name 和 city 两列，并将结果写入一个新

的 CSV 文件中。在 Azure Data Factory 中，我们可以使用映射数据流来实现这个

需求。

1. 创建数据流：在数据工厂中，选择“数据流”选项，创建一个新

的映射数据流。

2. 添加源：从左侧的工具栏中，将“源”拖到画布上，然后选择

CSV 文件作为源数据集。

3. 添加接收器：同样地，从工具栏中，将“接收器”拖到画布上，

选择 CSV 文件作为目标数据集。

运行后，新的 CSV 文件将只包含 name 和 city 两列的数据：

name, city

Alice, New York

Bob, Los Angeles

Charlie, Chicago

1.3 映射数据流的使用场景

映射数据流适用于以下几种场景：

� 数据清洗：去除数据中的空值、重复值、异常值等，以提高数据

质量。

� 数据转换：将数据从一种格式转换为另一种格式，如从 CSV 转换

工具，特别是在处理大规模数据时，其性能和效率优势更加明显。

2 数据集成工具：Azure Data Factory：使用映射数据流进行

数据转换

2.1 创建映射数据流

2.1.1 在 Azure Data Factory 中创建数据流

Azure Data Factory (ADF) 是一个云端数据整合服务，用于编排和自动化数据

集成工作流。映射数据流是 ADF 中用于数据转换的强大工具，它允许用户以图

形化界面设计数据转换逻辑，而无需编写复杂的代码。

2.1.1.1 步骤 1：创建数据流

1. 登录到 Azure Portal。

点，接收器是数据的终点。例如，你可以从 Azure SQL Database 中读取数据，

然后将其写入 Azure Blob Storage。

2.1.1.3 步骤 3：配置转换步骤

在源和接收器之间，你可以添加各种转换步骤，如选择、过滤、聚合等，

以处理和转换数据。

2.1.2 示例：使用映射数据流进行数据转换

假设我们有一个销售数据表，需要从 Azure SQL Database 中读取，然后进

行一些数据清洗和转换，最后将结果写入 Azure Blob Storage。

2.1.2.1 数据源：Azure SQL Database

表结构如下：

我们将转换后的数据以 CSV 格式存储。

2.1.2.3 数据转换逻辑

1. 选择特定列：只选择 ProductName、Quantity 和 Price。

2. 过滤：只保留 Quantity 大于 10 的记录。

3. 聚合：按 ProductName 分组，计算每个产品的总销售额。

2.1.2.4 实现步骤

1. 创建数据流：按照上述步骤创建一个新的映射数据流。

SELECT ProductName, Quantity, Price, SaleDate

FROM Sales

4. 添加转换：在源节点上，点击“+”，选择“Select”，然后选择

“Filter”和“Aggregate”。

5. 配置转换：

7. 配置接收器：设置 Blob Storage 连接，选择容器和文件路径，设

置文件格式为 CSV。

2.1.3 代码示例

虽然映射数据流主要通过图形界面操作，但下面是一个使用 ADF REST API

创建映射数据流的示例代码：

import requests

import json

url = "https://management.azure.com/subscriptions/{subscriptionId}/resourceGroups/{resource

GroupName}/providers/Microsoft.DataFactory/factories/{factoryName}/pipelines/{pipelineNam

data_flow = {

"name": "SalesDataTransformation",

"properties": {

"type": "MappingDataFlow",

"typeProperties": {

"sources": [

{

"dataset": {