使用Azure 数据工厂或Azure Synapse Analytics在Azure Blob 存储中复制和转换数据

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

本文概述了如何使用Azure 数据工厂和Azure Synapse管道中的复制活动从/向Azure Blob 存储复制数据。 它还介绍如何使用数据流活动转换Azure Blob 存储中的数据。 若要了解详细信息，请阅读 Azure 数据工厂 和 Azure Synapse Analytics 简介文章。

支持的功能

以下功能支持此Azure Blob 存储连接器：

（1） Azure集成运行时 （2） 自承载集成运行时

对于复制活动，此 Blob 存储连接器支持：

• 在常规用途的 Azure 存储帐户和热/冷 Blob 存储之间复制 Blob。
• 使用帐户密钥、服务共享访问签名（SAS）、服务主体或托管标识复制 blob，进行 Azure 资源身份验证。
• 从块 Blob、追加 Blob 或页 Blob 中复制数据，并仅将数据复制到块 Blob。
• 按原样复制 Blob，或者使用支持的文件格式和压缩编解码器分析或生成 Blob。
• 在复制期间保留文件元数据。

开始

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

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

使用 UI 创建Azure Blob 存储链接服务

使用以下步骤在Azure门户 UI 中创建Azure Blob 存储链接服务。

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

   • Azure 数据工厂
   • Azure Synapse

   

2. 搜索 blob 并选择Azure Blob 存储连接器。

   

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

   

连接器配置详细信息

以下部分详细介绍了用于定义特定于 Blob 存储的数据工厂和 Synapse 管道实体的属性。

链接的服务属性

此 Blob 存储连接器支持以下身份验证类型。 有关详细信息，请参阅相应部分。

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

匿名身份验证

Azure 数据工厂或 Synapse 管道中的存储帐户密钥身份验证支持以下属性：

示例：

{
    "name": "AzureBlobStorageAnonymous",
    "properties": {
        "annotations": [],
        "type": "AzureBlobStorage",
        "typeProperties": {
            "containerUri": "https://<accountname>.blob.core.chinacloudapi.cn/<containername>",
            "authenticationType": "Anonymous"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

UI 示例：

UI 体验如下图中所述。 此示例使用Azure打开的数据集作为源。

帐户密钥身份验证

Azure 数据工厂或 Synapse 管道中的存储帐户密钥身份验证支持以下属性：

示例：

{
    "name": "AzureBlobStorageLinkedService",
    "properties": {
        "type": "AzureBlobStorage",
        "typeProperties": {
          "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountname>;AccountKey=<accountkey>;EndpointSuffix=core.chinacloudapi.cn"
        },
        "connectVia": {
          "referenceName": "<name of Integration Runtime>",
          "type": "IntegrationRuntimeReference"
        }
    }
}

Example：将帐户密钥存储在 Azure 密钥保管库

{
    "name": "AzureBlobStorageLinkedService",
    "properties": {
        "type": "AzureBlobStorage",
        "typeProperties": {
            "connectionString": "DefaultEndpointsProtocol=https;AccountName=<accountname>;EndpointSuffix=core.chinacloudapi.cn",
            "accountKey": {
                "type": "AzureKeyVaultSecret",
                "store": {
                    "referenceName": "<Azure Key Vault linked service name>",
                    "type": "LinkedServiceReference"
                },
                "secretName": "<secretName>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

共享访问签名身份验证

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

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

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

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

示例：

{
    "name": "AzureBlobStorageLinkedService",
    "properties": {
        "type": "AzureBlobStorage",
        "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"
        }
    }
}

Example：将帐户密钥存储在 Azure 密钥保管库

{
    "name": "AzureBlobStorageLinkedService",
    "properties": {
        "type": "AzureBlobStorage",
        "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 更新链接服务。

服务主体身份验证

有关Azure 存储服务主体身份验证的一般信息，请参阅 Authenticate access to Azure 存储 using Microsoft Entra ID。

若要使用服务主体身份验证，请执行以下步骤：

1. 将应用程序注册到Microsoft 标识平台。 若要了解如何操作，请参阅 Quickstart：向 Microsoft 标识平台 注册应用程序。 记下以下值，这些值用于定义链接服务：

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

2. 在Azure Blob 存储中授予服务主体适当的权限。 有关角色的详细信息，请参阅 使用 Azure 门户分配 Azure 角色以访问 blob 和队列数据。

   • 作为来源，在“访问控制 (IAM)”中至少授予“存储 Blob 数据读取者”角色。
   • 作为 接收器，在“访问控制 (IAM)”中至少授予“Storage Blob Data Contributor (存储 Blob 数据参与者)”角色。

Azure Blob 存储链接服务支持这些属性：

示例：

{
    "name": "AzureBlobStorageLinkedService",
    "properties": {
        "type": "AzureBlobStorage",
        "typeProperties": {            
            "serviceEndpoint": "https://<accountName>.blob.core.chinacloudapi.cn/",
            "accountKind": "StorageV2",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalKey": {
                "type": "SecureString",
                "value": "<service principal key>"
            },
            "tenant": "<tenant info, e.g. microsoft.partner.onmschina.cn>" 
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

系统分配的托管身份验证

数据工厂或 Synapse 管道可与用于 Azure 资源的系统分配的托管标识相关联，该标识表示该资源用于向其他 Azure 服务进行身份验证。 可以像使用自己的服务主体一样，直接使用此系统分配的托管标识进行 Blob 存储身份验证。 这个指定的资源可通过此方法访问和复制 Blob 存储中的数据或将数据复制到其中。 若要了解有关Azure资源的托管标识的详细信息，请参阅 Azure 资源的托管标识

有关 Azure 存储身份验证的一般信息，请参阅 使用 Microsoft Entra ID 验证对 Azure 存储的访问。 若要使用托管标识进行Azure资源身份验证，请执行以下步骤：

1. 检索系统分配的托管标识信息，方法是复制随工厂或 Synapse 工作区生成的系统分配的托管标识对象 ID 的值。

2. 在 Azure Blob 存储 中授予托管身份权限。 有关角色的详细信息，请参阅 使用 Azure 门户分配 Azure 角色以访问 blob 和队列数据。

   • 作为来源，在“访问控制 (IAM)”中至少授予“存储 Blob 数据读取者”角色。
   • 作为 接收器，在“访问控制 (IAM)”中至少授予“Storage Blob Data Contributor (存储 Blob 数据参与者)”角色。

Azure Blob 存储链接服务支持这些属性：

示例：

{
    "name": "AzureBlobStorageLinkedService",
    "properties": {
        "type": "AzureBlobStorage",
        "typeProperties": {            
            "serviceEndpoint": "https://<accountName>.blob.core.chinacloudapi.cn/",
            "accountKind": "StorageV2" 
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

用户指定的托管标识身份验证

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

有关 Azure 存储身份验证的一般信息，请参阅 使用 Microsoft Entra ID 对 Azure 存储进行身份验证。 若要使用用户分配的托管标识身份验证，请执行以下步骤：

1. 创建一个或多个用户分配的托管标识并在Azure Blob 存储中授予权限。 有关角色的详细信息，请参阅 使用 Azure 门户分配 Azure 角色以访问 blob 和队列数据。

   • 作为来源，在“访问控制 (IAM)”中至少授予“存储 Blob 数据读取者”角色。
   • 作为 接收器，在“访问控制 (IAM)”中至少授予“Storage Blob Data Contributor (存储 Blob 数据参与者)”角色。

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

Azure Blob 存储链接服务支持这些属性：

示例：

{
    "name": "AzureBlobStorageLinkedService",
    "properties": {
        "type": "AzureBlobStorage",
        "typeProperties": {            
            "serviceEndpoint": "https://<accountName>.blob.core.chinacloudapi.cn/",
            "accountKind": "StorageV2",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

数据集属性

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

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

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

在基于格式的数据集中的 location 设置下，Azure Blob 存储支持以下属性：

示例：

{
    "name": "DelimitedTextDataset",
    "properties": {
        "type": "DelimitedText",
        "linkedServiceName": {
            "referenceName": "<Azure Blob Storage linked service name>",
            "type": "LinkedServiceReference"
        },
        "schema": [ < physical schema, optional, auto retrieved during authoring > ],
        "typeProperties": {
            "location": {
                "type": "AzureBlobStorageLocation",
                "container": "containername",
                "folderPath": "folder/subfolder"
            },
            "columnDelimiter": ",",
            "quoteChar": "\"",
            "firstRowAsHeader": true,
            "compressionCodec": "gzip"
        }
    }
}

复制活动 属性

有关可用于定义活动的各部分和属性的完整列表，请参阅管道一文。 本部分提供 Blob 存储源和接收器支持的属性的列表。

将 Blob 存储用作源类型

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

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

在基于格式的复制源中的 storeSettings 设置下，Azure Blob 存储支持以下属性：

示例：

"activities":[
    {
        "name": "CopyFromBlob",
        "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": "AzureBlobStorageReadSettings",
                    "recursive": true,
                    "wildcardFolderPath": "myfolder*A",
                    "wildcardFileName": "*.csv"
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

用作接收器类型的 Blob 存储

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

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

在基于格式的复制接收器中的 storeSettings 设置下，Azure Blob 存储支持以下属性：

示例：

"activities":[
    {
        "name": "CopyFromBlob",
        "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": "AzureBlobStorageWriteSettings",
                    "copyBehavior": "PreserveHierarchy",
                    "metadata": [
                        {
                            "name": "testKey1",
                            "value": "value1"
                        },
                        {
                            "name": "testKey2",
                            "value": "value2"
                        },
                        {
                            "name": "lastModifiedKey",
                            "value": "$$LASTMODIFIED"
                        }
                    ]
                }
            }
        }
    }
]

文件夹和文件筛选器示例

本部分介绍使用通配符筛选器生成文件夹路径和文件名的行为。

文件列表示例

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

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

一些递归和复制行为示例

此部分描述了在不同的 recursive 和 copyBehavior 值组合下，复制操作的结果行为。

在复制期间保留元数据

将文件从 Amazon S3、Azure Blob 存储或Azure Data Lake Storage Gen2复制到Azure Data Lake Storage Gen2或Azure Blob 存储时，可以选择保留文件元数据和数据。 从保留元数据中了解更多信息。

映射数据流属性

在映射数据流中转换数据时，可以按以下格式从Azure Blob 存储读取和写入文件：

• Avro
• 带分隔符的文本
• Delta
• Excel
• JSON
• Parquet

该格式的特定设置可在其文档中找到。 有关详细信息，请参阅映射数据流中的源转换和映射数据流中的汇转换。

源转换

在源转换中，可以从容器、文件夹或Azure Blob 存储中的单个文件进行读取。 使用“源选项”选项卡，可以管理文件的读取方式。

通配符路径：如果使用通配符模式，则会指示服务在单个源转换中循环访问每个匹配的文件夹和文件。 这是在单个流中处理多个文件的有效方法。 使用将鼠标指针悬停在现有通配符模式上时出现的加号来添加多个通配符匹配模式。

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

通配符示例：

• * 表示任意字符集。

• ** 表示递归目录嵌套。

• ? 替换一个字符。

• [] 匹配括号内的一个或多个字符。

• /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 文件。

分区根路径：如果文件源中存在 格式（例如 key=value）的分区文件夹，则可以向该分区文件夹树的顶层分配数据流中的列名称。

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

使用分区根路径 设置来定义文件夹结构的顶级是什么。 通过数据预览查看数据内容时会看到，该服务会添加在每个文件夹级别中找到的已解析的分区。

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

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

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

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

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

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

可以将“源”指定为：

/data/sales

可以将“为”指定为：

/backup/priorSales

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

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

启用变更数据捕获：如果为 true，则只能从上次运行中获取新的或更改后的文件。 将始终从首次运行中获取完整快照数据的初始加载，然后仅在下次运行中捕获新的或更改后的文件。

接收器属性

在接收器转换中，可以写入Azure Blob 存储中的容器或文件夹。 使用“设置”选项卡，可以管理文件的写入方式。

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

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

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

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

Lookup 操作属性

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

GetMetadata 活动属性

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

删除活动属性

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

遗留模型

旧数据集模型

示例：

{
    "name": "AzureBlobDataset",
    "properties": {
        "type": "AzureBlob",
        "linkedServiceName": {
            "referenceName": "<Azure Blob Storage linked service name>",
            "type": "LinkedServiceReference"
        },
        "typeProperties": {
            "folderPath": "mycontainer/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": "CopyFromBlob",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Azure Blob input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "BlobSource",
                "recursive": true
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

复制活动的旧接收器模型

示例：

"activities":[
    {
        "name": "CopyToBlob",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Azure Blob output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "BlobSink",
                "copyBehavior": "PreserveHierarchy"
            }
        }
    }
]

更改数据捕获

Azure 数据工厂只有在映射数据流源转换中启用“启用更改数据捕获”，才能从Azure Blob 存储获取新的或更改的文件。 使用此连接器选项，可以仅读取新的或更新后的文件，并在将转换后的数据加载到所选目标数据集之前应用转换。 有关详细信息，请参阅 变更数据捕获 。

相关内容

关于 复制活动 作为源和汇支持的数据存储列表，请参阅 支持的数据存储。