使用 Azure 数据工厂或 Azure Synapse Analytics 复制和转换 Azure Data Lake Storage Gen2 中的数据

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

Azure Data Lake Storage Gen2 (ADLS Gen2) 是一组专用于大数据分析的功能，内置于 Azure Blob 存储中。 它可使用文件系统和对象存储范例与数据进行交互。

本文概述了如何使用复制活动从/向 Azure Data Lake Storage Gen2 复制数据，并使用数据流在 Azure Data Lake Storage Gen2 中转换数据。 有关详细信息，请阅读 Azure 数据工厂 或 Azure Synapse Analytics 的简介文章。

支持的功能

此 Azure Data Lake Storage Gen2 连接器支持以下功能：

① Azure 集成运行时 ② 自承载集成运行时

对于复制活动，使用此连接器可以：

• 通过使用帐户密钥、服务主体或 Azure 资源托管标识进行 Azure 资源身份验证来从/向 Azure Data Lake Storage Gen2 复制数据。
• 按原样复制文件，或使用支持的文件格式和压缩编解码器解析/生成文件。
• 在复制期间保留文件元数据。

开始

若要使用管道执行复制活动，可以使用以下工具或 SDK 之一：

• 复制数据工具
• Azure 门户
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure 资源管理器 模板

使用 UI 创建 Azure Data Lake Storage Gen2 链接服务

使用以下步骤在 Azure 门户 UI 中创建 Azure Data Lake Storage Gen2 链接服务。

1. 浏览到 Azure 数据工厂或 Synapse 工作区中的“管理”选项卡，并选择“链接服务”，然后单击“新建”：

   • Azure 数据工厂
   • Azure Synapse

   

2. 搜索“Azure Data Lake Storage Gen2”并选择“Azure Data Lake Storage Gen2 连接器”。

   

3. 配置服务详细信息、测试连接并创建新的链接服务。

   

连接器配置详细信息

下面各部分介绍了用于定义特定于 Data Lake Storage Gen2 的数据工厂和 Synapse 管道实体的属性。

链接服务属性

Azure Data Lake Storage Gen2 连接器支持以下身份验证类型。 请参阅相应部分，了解详细信息：

• 帐户密钥验证
• 共享访问签名身份验证
• 服务主体身份验证
• 系统分配的托管标识符身份验证
• 用户分配的托管标识身份验证

帐户密钥身份验证

若要使用存储帐户密钥身份验证，支持以下属性：

示例：

{
    "name": "AzureDataLakeStorageGen2LinkedService",
    "properties": {
        "type": "AzureBlobFS",
        "typeProperties": {
            "url": "https://<accountname>.dfs.core.chinacloudapi.cn", 
            "accountkey": { 
                "type": "SecureString", 
                "value": "<accountkey>" 
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

共享访问签名身份验证

共享访问签名对存储帐户中的资源提供委托访问。 使用共享访问签名可以在指定的时间内授予客户端对存储帐户中对象的有限访问权限。

无需共享帐户访问密钥。 共享访问签名是一个 URI，在其查询参数中包含对存储资源已验证访问所需的所有信息。 若要使用共享访问签名访问存储资源，客户端只需将共享访问签名传入到相应的构造函数或方法。

有关共享访问签名的详细信息，请参阅共享访问签名：了解共享访问签名模型。

以下属性支持使用共享访问签名身份验证：

示例：

{
    "name": "AzureDataLakeStorageGen2LinkedService",
    "properties": {
        "type": "AzureBlobFS",
        "typeProperties": {
            "sasUri": {
                "type": "SecureString",
                "value": "<SAS URI of the Azure Storage resource e.g. https://<accountname>.blob.core.chinacloudapi.cn/?sv=<storage version>&st=<start time>&se=<expire time>&sr=<resource>&sp=<permissions>&sip=<ip range>&spr=<protocol>&sig=<signature>>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

示例：在 Azure 密钥保管库 中存储帐户密钥

{
    "name": "AzureDataLakeStorageGen2LinkedService",
    "properties": {
        "type": "AzureBlobFS",
        "typeProperties": {
            "sasUri": {
                "type": "SecureString",
                "value": "<SAS URI of the Azure Storage resource without token e.g. https://<accountname>.blob.core.chinacloudapi.cn/>"
            },
            "sasToken": {
                "type": "AzureKeyVaultSecret",
                "store": {
                    "referenceName": "<Azure Key Vault linked service name>", 
                    "type": "LinkedServiceReference"
                },
                "secretName": "<secretName with value of SAS token e.g. ?sv=<storage version>&st=<start time>&se=<expire time>&sr=<resource>&sp=<permissions>&sip=<ip range>&spr=<protocol>&sig=<signature>>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

在创建共享访问签名 URI 时，请注意以下几点：

• 根据链接服务（读取、写入、读/写）的用法，设置针对对象的适当读/写权限。
• 根据需要设置到期时间。 确保存储对象的访问权限不会在管道的活动期限内过期。
• 应该根据需要在正确的容器或 Blob 中创建 URI。 数据工厂或 Synapse 管道可以使用 Blob 的共享访问签名 URI 访问该特定 blob。 数据工厂或 Synapse 管道可以使用 Blob 存储容器的共享访问签名 URI 迭代该容器中的 blob。 以后若要提供更多/更少对象的访问权限或需要更新共享访问签名 URI，请记得使用新 URI 更新链接服务。

服务主体身份验证

若要使用服务主体身份验证，请按照以下步骤操作。

1. 将应用程序注册到 Microsoft 标识平台。 要了解具体操作，请参阅 快速入门：使用 Microsoft 标识平台注册应用程序。 记下以下值，这些值用于定义链接服务：

   • 应用 ID
   • 应用程序密钥
   • 租户 ID

2. 向服务主体对象授予适当的权限。 请参阅文件和目录上的访问控制列表中的示例，了解 Data Lake Storage Gen2 中权限的工作原理。

   • 作为源：在存储资源管理器中，至少授予对所有上游文件夹和文件系统的“执行”权限，并授予对文件的“读取”权限以进行复制 。 或者，在访问控制 (IAM) 中，授予至少 存储 Blob 数据读取者 角色。
   • 作为接收器：在存储资源管理器中，至少授予所有上游文件夹和文件系统的“执行”权限，以及接收器文件夹的“写入”权限。 或者，在访问控制 (IAM) 中，至少授予 Storage Blob Data Contributor 角色。

链接服务支持以下属性：

示例：使用服务主体密钥身份验证

也可以将服务主体密钥存储在 Azure 密钥保管库 中。

{
    "name": "AzureDataLakeStorageGen2LinkedService",
    "properties": {
        "type": "AzureBlobFS",
        "typeProperties": {
            "url": "https://<accountname>.dfs.core.chinacloudapi.cn", 
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredentialType": "ServicePrincipalKey",
            "servicePrincipalCredential": {
                "type": "SecureString",
                "value": "<service principal key>"
            },
            "tenant": "<tenant info, e.g. microsoft.partner.onmschina.cn>" 
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

示例：使用服务主体证书身份验证

{
    "name": "AzureDataLakeStorageGen2LinkedService",
    "properties": {
        "type": "AzureBlobFS",
        "typeProperties": {
            "url": "https://<accountname>.dfs.core.chinacloudapi.cn", 
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredentialType": "ServicePrincipalCert",
            "servicePrincipalCredential": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<AKV reference>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<certificate name in AKV>" 
            },
            "tenant": "<tenant info, e.g. microsoft.partner.onmschina.cn>" 
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

系统分配的托管身份验证

数据工厂或 Synapse 工作区可与系统分配的托管标识相关联。 你可以像使用自己的服务主体一样，直接使用此系统分配的托管标识进行 Data Lake Storage Gen2 身份验证。 这样，此指定的数据工厂或工作区就可以访问并向/从你的 Data Lake Storage Gen2 复制数据。

若要使用系统分配的托管标识身份验证，请执行以下步骤。

1. 通过复制随数据工厂或 Synapse 工作区一起生成的托管标识对象 ID 的值，检索系统分配的托管标识信息。

2. 向系统分配的托管标识授予适当权限。 请参阅文件和目录上的访问控制列表中的示例，了解 Data Lake Storage Gen2 中权限的工作方式。

   • 作为源：在存储资源管理器中，至少授予对所有上游文件夹和文件系统的“执行”权限，并授予对文件的“读取”权限以进行复制 。 或者，在访问控制 (IAM) 中，授予至少 “存储 Blob 数据读取者” 角色。
   • 作为接收端：在存储资源管理器中，至少授予所有上游文件夹和文件系统的执行权限，以及接收端文件夹的写入权限。 或者，在访问控制 (IAM) 中，授予至少 Storage Blob Data Contributor 角色。

链接服务支持以下属性：

示例：

{
    "name": "AzureDataLakeStorageGen2LinkedService",
    "properties": {
        "type": "AzureBlobFS",
        "typeProperties": {
            "url": "https://<accountname>.dfs.core.chinacloudapi.cn", 
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

用户分配的托管身份验证

可以为数据工厂分配一个或多个用户分配的托管标识。 可以将此用户分配的托管标识用于 Blob 存储身份验证，该身份验证允许访问 Data Lake Storage Gen2 中的数据，或从中复制数据及将数据复制到其中。 若要详细了解 Azure 资源托管标识，请参阅 Azure 资源托管标识

若要使用用户分配的托管标识身份验证，请执行以下步骤：

1. 创建一个或多个用户分配的托管标识并授予对 Azure Data Lake Storage Gen2 的访问权限。 通过 文件和目录上的访问控制列表中的一些示例，了解 Data Lake Storage Gen2 中权限的工作方式。

   • 作为源：在存储资源管理器中，至少授予对所有上游文件夹和文件系统的“执行”权限，并授予对文件的“读取”权限以进行复制 。 或者，在访问控制 (IAM) 中，授予至少 “存储 Blob 数据读取者” 角色。
   • 作为接收端：在 存储资源管理器 中，至少授予所有上游文件夹和文件系统的执行权限，以及接收端文件夹的写入权限。 或者，在访问控制 (IAM) 中，授予至少 Storage Blob Data Contributor 角色。

2. 为数据工厂分配一个或多个用户分配的托管标识，并为每个用户分配的托管标识创建凭据。

链接服务支持以下属性：

示例：

{
    "name": "AzureDataLakeStorageGen2LinkedService",
    "properties": {
        "type": "AzureBlobFS",
        "typeProperties": {
            "url": "https://<accountname>.dfs.core.chinacloudapi.cn", 
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
                }
            },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

数据集属性

有关可用于定义数据集的各个部分和属性的完整列表，请参阅数据集。

Azure 数据工厂支持以下文件格式。 请参阅每篇文章，了解基于格式的设置。

• Avro 格式
• 二进制格式
• 带分隔符的文本格式
• Excel 格式
• Iceberg 格式
• JSON 格式
• ORC 格式
• Parquet 格式
• XML 格式

在基于格式的数据集中的 location 设置下，Data Lake Storage Gen2 支持以下属性：

示例：

{
    "name": "DelimitedTextDataset",
    "properties": {
        "type": "DelimitedText",
        "linkedServiceName": {
            "referenceName": "<Data Lake Storage Gen2 linked service name>",
            "type": "LinkedServiceReference"
        },
        "schema": [ < physical schema, optional, auto retrieved during authoring > ],
        "typeProperties": {
            "location": {
                "type": "AzureBlobFSLocation",
                "fileSystem": "filesystemname",
                "folderPath": "folder/subfolder"
            },
            "columnDelimiter": ",",
            "quoteChar": "\"",
            "firstRowAsHeader": true,
            "compressionCodec": "gzip"
        }
    }
}

复制活动属性

要完整了解可供活动定义使用的各个部分和属性，请参阅复制活动配置和管道和活动。 本部分列出了 Data Lake Storage Gen2 源和接收器支持的属性。

Azure Data Lake Storage Gen2 作为源类型

Azure 数据工厂支持以下文件格式。 请参阅有关基于格式的设置的每篇文章。

• Avro 格式
• 二进制格式
• 带分隔符的文本格式
• Excel 格式
• JSON 格式
• ORC 格式
• Parquet 格式
• XML 格式

从 ADLS Gen2 复制数据的方式有多种：

• 从数据集中指定的给定路径复制。
• 有关针对文件夹路径或文件名的通配符筛选器，请参阅 wildcardFolderPath 和 wildcardFileName。
• 将给定文本文件中定义的文件作为文件集复制（请参阅 fileListPath）。

在基于格式的复制源中的 storeSettings 设置下，以下属性可用于 Data Lake Storage Gen2：

示例：

"activities":[
    {
        "name": "CopyFromADLSGen2",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Delimited text input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "DelimitedTextSource",
                "formatSettings":{
                    "type": "DelimitedTextReadSettings",
                    "skipLineCount": 10
                },
                "storeSettings":{
                    "type": "AzureBlobFSReadSettings",
                    "recursive": true,
                    "wildcardFolderPath": "myfolder*A",
                    "wildcardFileName": "*.csv"
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

Azure Data Lake Storage Gen2 作为接收器类型

Azure 数据工厂支持以下文件格式。 请参阅每篇文章中的基于格式的设置。

• Avro 格式
• 二进制格式
• 带分隔符的文本格式
• Iceberg 格式
• JSON 格式
• ORC 格式
• Parquet 格式

在基于格式的复制接收器中的 storeSettings 设置下，Data Lake Storage Gen2 支持以下属性：

示例：

"activities":[
    {
        "name": "CopyToADLSGen2",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Parquet output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "ParquetSink",
                "storeSettings":{
                    "type": "AzureBlobFSWriteSettings",
                    "copyBehavior": "PreserveHierarchy",
                    "metadata": [
                        {
                            "name": "testKey1",
                            "value": "value1"
                        },
                        {
                            "name": "testKey2",
                            "value": "value2"
                        },
                        {
                            "name": "lastModifiedKey",
                            "value": "$$LASTMODIFIED"
                        }
                    ]
                }
            }
        }
    }
]

文件夹和文件筛选器示例

本部分介绍文件夹路径和文件名在使用通配符筛选器时的表现。

文件列表示例

本部分介绍了在复制活动源中使用文件列表路径时的结果行为。

假设有以下源文件夹结构，并且要复制加粗显示的文件：

一些 recursive 和 copyBehavior 示例

本节介绍了将 recursive 和 copyBehavior 值进行不同组合所产生的复制操作行为。

在复制期间保留元数据

将文件从 Amazon S3/Azure Blob/Azure Data Lake Storage Gen2 复制到 Azure Data Lake Storage Gen2/Azure Blob 时，可以选择在复制数据的同时保留文件元数据。 详细了解保留元数据。

映射数据流属性

在映射数据流中转换数据时，可以在 Azure Data Lake Storage Gen2 中读取和写入以下格式的文件：

• Avro
• 通用数据模型
• 带分隔符的文本
• Delta
• Excel
• JSON
• Parquet

格式特定的设置位于该格式的文档中。 有关详细信息，请参阅映射数据流中的源转换和映射数据流中的接收器转换。

源转换

在源转换中，可以从 Azure Data Lake Storage Gen2 中的容器、文件夹或单个文件进行读取。 通过“源选项”选项卡，可以管理如何读取文件。

通配符路径： 如果使用通配符模式，则是指示 ADF 在单个“源转换”中遍历每个匹配的文件夹和文件。 这是在单个流中处理多个文件的有效方法。 使用将鼠标悬停在现有通配符模式上时出现的“+”号来添加多个通配符匹配模式。

从源容器中，选择与模式匹配的一系列文件。 在该数据集中只能指定一个容器。 因此，通配符路径也必须包含从根文件夹开始的文件夹路径。

通配符示例：

• * 表示任意一组字符

• ** 表示递归目录嵌套

• ? 替换一个字符

• [] 与括号中的一个或多个字符匹配

• /data/sales/**/*.csv 获取 /data/sales 下的所有 csv 文件

• /data/sales/20??/**/ 获取 20 世纪的所有文件

• /data/sales/*/*/*.csv 获取 /data/sales 下两级目录中的 csv 文件

• /data/sales/2004/*/12/[XY]1?.csv 获取 2004 年 12 月所有以前缀为 X 或 Y 的两位数字开头的 csv 文件

分区根路径：如果文件源中存在采用 格式（例如 year=2019）的分区文件夹，则可以将该分区文件夹树的顶层映射到数据流中的某个列名。

首先，设置一个通配符，用于包含作为分区文件夹以及要读取的叶文件的所有路径。

使用“分区根路径”设置来定义文件夹结构的顶级。 通过数据预览查看数据内容时会看到，ADF 会添加在每个文件夹级别中找到的已识别的分区。

文件列表： 这是一个文件集。 创建一个文本文件，其中包含要处理的相对路径文件的列表。 指向此文本文件。

用于存储文件名的列： 将源文件的名称存储在数据的列中。 请在此处输入新列名称以存储文件名字符串。

完成后： 数据流运行后，可以选择不对源文件执行任何操作、删除源文件或移动源文件。 用于移动的路径是相对的。

要将源文件在后期处理后移到其他位置，请首先选择文件操作为“移动”。 然后，设置“from”目录。 如果未对路径使用任何通配符，则“from”设置将与源文件夹相同。

如果源路径包含通配符，则语法将如下所示：

/data/sales/20??/**/*.csv

可将“from”指定为

/data/sales

以及“到”作为

/backup/priorSales

在此例中，源自 /data/sales 下的所有文件都将移动到 /backup/priorSales。

按上次修改时间筛选： 通过指定上次修改的日期范围，可以筛选要处理的文件。 所有日期时间均采用 UTC 时间。

启用变更数据捕获：如果为 true，则只能获取自上次运行以来新增或更改的文件。 首次运行时始终会先加载完整快照数据，后续运行仅捕获新增或更改的文件。 有关详细信息，请参阅变更数据捕获。

接收端属性

在接收器转换中，可以写入到 Azure Data Lake Storage Gen2 中的容器或文件夹。 通过设置选项卡，可以管理文件的写入方式。

清除文件夹： 确定在写入数据之前是否清除目标文件夹。

文件名选项： 确定目标文件在目标文件夹中的命名方式。 文件名选项有：

• 默认：允许 Spark 根据 PART 默认值为文件命名。
• 模式：输入一种用于按分区为输出文件编号的模式。 例如，loans[n].csv 将创建 loans1.csv、loans2.csv 等等。
• 每个分区：为每个分区输入一个文件名。
• 按列中的数据：将输出文件设置为列的值。 此路径是相对于数据集容器而非目标文件夹的。 如果数据集中有文件夹路径，则会将其覆盖。
• 输出到一个文件：将分区输出文件合并为一个命名的文件。 此路径是数据集文件夹的相对路径。 请注意，合并操作可能会因为节点大小而失败。 对于大型数据集，不建议使用此选项。

全部引用： 确定是否将所有值括在引号中

umask

可以选择使用 POSIX 读取、写入、执行所有者、用户和组的标志来设置 umask 文件。

预处理命令和后处理命令

可以选择在写入 ADLS Gen2 接收器之前或之后执行 Hadoop 文件系统命令。 支持以下命令：

• cp
• mv
• rm
• mkdir

示例：

• mkdir /folder1
• mkdir -p folder1
• mv /folder1/*.* /folder2/
• cp /folder1/file1.txt /folder2
• rm -r /folder1

表达式生成器也支持参数，例如：

mkdir -p {$tempPath}/commands/c1/c2 mv {$tempPath}/commands/*.* {$tempPath}/commands/c1/c2

默认情况下，文件夹将在 user/root 下创建。 请参阅以“/”表示的顶级容器。

查找活动属性

若要了解有关属性的详细信息，请查看 Lookup 活动。

GetMetadata 活动属性

若要了解有关属性的详细信息，请查看 GetMetadata 活动

删除活动属性

若要详细了解这些属性，请查看 删除活动

旧版模型

旧数据集模型

示例：

{
    "name": "ADLSGen2Dataset",
    "properties": {
        "type": "AzureBlobFSFile",
        "linkedServiceName": {
            "referenceName": "<Azure Data Lake Storage Gen2 linked service name>",
            "type": "LinkedServiceReference"
        },
        "typeProperties": {
            "folderPath": "myfilesystem/myfolder",
            "fileName": "*",
            "modifiedDatetimeStart": "2018-12-01T05:00:00Z",
            "modifiedDatetimeEnd": "2018-12-01T06:00:00Z",
            "format": {
                "type": "TextFormat",
                "columnDelimiter": ",",
                "rowDelimiter": "\n"
            },
            "compression": {
                "type": "GZip",
                "level": "Optimal"
            }
        }
    }
}

复制活动的旧版源模型

示例：

"activities":[
    {
        "name": "CopyFromADLSGen2",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<ADLS Gen2 input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzureBlobFSSource",
                "recursive": true
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

旧复制活动接收槽模型

示例：

"activities":[
    {
        "name": "CopyToADLSGen2",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<ADLS Gen2 output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzureBlobFSSink",
                "copyBehavior": "PreserveHierarchy"
            }
        }
    }
]

变更数据捕获

Azure 数据工厂只能从 Azure Data Lake Storage Gen2 获取新的或更改后的文件，方法是在映射数据流源转换中启用“启用变更数据捕获”。 使用此连接器选项，可以仅读取新的或更新后的文件，并在将转换后的数据加载到所选目标数据集之前应用转换。

请确保管道和活动名称保持不变，以便始终可以从上次运行中记录检查点，从而从该检查点获取更改。 如果更改管道名称或活动名称，检查点将重置，并且你将在下一次运行中从头开始。

调试管道时，启用变更数据捕获也正常工作。 在调试运行期间刷新浏览器时，检查点将重置。 对调试运行的结果感到满意后，可以发布和触发管道。 无论调试运行记录的上一个检查点是什么，它都将始终从头开始。

在监视部分，你始终有机会重新运行流水线。 在这样做时，将始终从所选管道运行中的检查点记录获取更改。

相关内容

有关复制活动支持作为源和目标的数据存储的列表，请参阅受支持的数据存储。