Azure 数据工厂和 Azure Synapse Analytics 中的复制活动

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

在 Azure 数据工厂和 Synapse 管道中，可使用复制活动在本地与云数据存储之间复制数据。 复制数据后，可以使用其他活动进一步转换和分析数据。 还可使用复制活动发布有关商业智能 (BI) 和应用程序消耗的转换和分析结果。

复制活动在集成运行时上执行。 对于不同的数据复制方案，你可以使用不同类型的集成运行时：

• 当你在两个可通过互联网从任何 IP 公开访问的数据存储之间复制数据时，可以在复制活动中使用 Azure Integration Runtime。 此集成运行时较为安全可靠、可缩放并全局可用。
• 当你在本地数据存储或位于带有访问控制的网络（例如 Azure 虚拟网络）中的数据存储之间进行数据复制时，需要设置一个自承载集成运行时。

集成运行时需要与每个源数据存储和接收器数据存储相关联。 有关复制活动如何确定要使用的集成运行时的信息，请参阅确定要使用哪个 IR。

若要将数据从源复制到接收器，运行复制活动的服务将执行以下步骤：

1. 读取源数据存储中的数据。
2. 执行序列化/反序列化、压缩/解压缩、列映射，等等。 此服务基于输入数据集、输出数据集和复制活动的配置执行这些操作。
3. 将数据写入接收器/目标数据存储。

支持的数据存储和格式

支持的文件格式

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

• Avro 格式
• 二进制格式
• 带分隔符的文本格式
• Excel 格式
• Iceberg 格式（仅适用于 Azure Data Lake Storage Gen2）
• JSON 格式
• ORC 格式
• Parquet 格式
• XML 格式

可以使用复制活动来复制基于文件的两个数据存储间的文件，这种情况下，数据可以高效复制而无需任何序列化或反序列化。 此外，还可以分析或生成给定格式的文件。例如，你可以执行以下操作：

• 从 SQL Server 数据库复制数据，并将数据以 Parquet 格式写入 Azure Data Lake Storage Gen2。
• 从本地文件系统中复制文本 (CSV) 格式文件，并将其以 Avro 格式写入 Azure Blob 存储。
• 从本地文件系统复制压缩文件，动态解压缩它们，然后将提取的文件写入 Azure Data Lake Storage Gen2。
• 从 Azure Blob 存储复制 Gzip 压缩文本 (CSV) 格式的数据，并将其写入 Azure SQL 数据库。
• 需要序列化/反序列化或压缩/解压缩的许多其他活动。

支持的区域

为复制活动提供支持的服务面向 Azure Integration Runtime 位置中列出的区域和地理位置全球发布。 全局可用拓扑可确保高效的数据移动，此类移动通常避免跨区域跃点。 请参阅 “按区域分类的产品 ”，以检查特定区域中数据工厂、Synapse 工作区和数据移动的可用性。

配置

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

• 复制数据工具
• Azure 门户
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure Resource Manager 模板

通常，若要使用 Azure 数据工厂或 Synapse 管道中的复制活动，需要执行以下操作：

1. 创建用于源数据存储和接收器数据存储的链接服务。 你可以在本文的受支持的数据存储和格式部分找到受支持的连接器列表。 有关配置信息和支持的属性，请参阅连接器文章的“链接服务属性”部分。
2. 为源和接收器创建数据集。 有关配置信息和支持的属性，请参阅源和接收器连接器文章的“数据集属性”部分。
3. 创建包含复制活动的管道。 接下来的部分将提供示例。

语法

以下复制活动模板包含受支持属性的完整列表。 指定适合你的方案的属性。

"activities":[
    {
        "name": "CopyActivityTemplate",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<source dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<sink dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>",
                <properties>
            },
            "sink": {
                "type": "<sink type>"
                <properties>
            },
            "translator":
            {
                "type": "TabularTranslator",
                "columnMappings": "<column mapping>"
            },
            "dataIntegrationUnits": <number>,
            "parallelCopies": <number>,
            "enableStaging": true/false,
            "stagingSettings": {
                <properties>
            },
            "enableSkipIncompatibleRow": true/false,
            "redirectIncompatibleRowSettings": {
                <properties>
            }
        }
    }
]

语法详细信息

监视

可以编程方式直观地监视在 Azure 数据工厂和 Synapse 管道中运行的复制活动。 有关详细信息，请参阅监视复制活动。

增量复制

通过数据工厂和 Synapse 管道，可以递增方式将增量数据从源数据存储复制到接收器数据存储。 有关详细信息，请参阅教程：以增量方式复制数据。

性能和优化

复制活动监视体验显示每个活动运行的复制性能统计信息。 复制活动性能和可伸缩性指南介绍通过复制活动影响数据移动性能的关键因素。 其中还列出了在测试期间观测到的性能值，并介绍了如何优化复制活动的性能。

从上次失败的运行继续

当你在基于文件的存储之间，以二进制格式原样复制大文件，并选择从源到目标保留文件夹/文件层次结构时，复制活动支持从上一次失败的运行继续。例如，在将数据从 Amazon S3 迁移到 Azure Data Lake Storage Gen2 时。 它适用于以下基于文件的连接器：Amazon S3、Amazon S3 兼容存储、Azure Blob、Azure Data Lake Storage Gen1、Azure Data Lake Storage Gen2、Azure 文件、文件系统、FTP、Google 云存储、HDFS、Oracle 云存储 和 SFTP。

可以通过以下两种方式使用复制活动继续操作：

• 活动级重试：你可以设置复制活动的重试计数。 在管道执行期间，如果此复制活动运行失败，则下一次自动重试从上次试用的失败点开始。

• 从失败的活动重新运行：管道执行完成后，还可以在 ADF UI 监视视图中或以编程方式触发从失败的活动重新运行。 如果失败的活动是复制活动，则管道不仅会从此活动重新运行，而且还会从上一次运行的故障点继续。

  

要注意的几点：

• 继续操作发生在文件级别。 如果复制活动在复制文件时失败，下一次运行时会重新复制此特定文件。
• 为了让继续功能正常工作，在重新运行之间不要更改复制活动的设置。
• 从 Amazon S3、Azure Blob、Azure Data Lake Storage Gen2 和 Google Cloud Storage 复制数据时，复制活动可以从任意数量已复制的文件恢复。 然而，对于其余以基于文件的连接器作为源的情况，目前复制活动仅支持从有限数量的文件继续，通常在数万级别，并且会因文件路径的长度而有所不同；超过该数量的文件在重新运行时将被重新复制。

对于二进制文件复制以外的其他情况，复制活动将从头开始重新运行。

随数据保留元数据

在将数据从源复制到目标的过程中，例如在数据湖迁移场景中，你还可以选择使用复制活动在复制数据的同时保留元数据和 ACL。 有关详细信息，请参阅保留元数据。

将元数据标记添加到基于文件的接收器

当接收器基于 Azure 存储（Azure 数据湖存储或 Azure Blob 存储）时，我们可以选择向文件添加一些元数据。 这些元数据将作为键值对显示为文件属性的一部分。 对于所有类型的基于文件的接收器，可以使用管道参数、系统变量、函数和变量添加涉及动态内容的元数据。 除此之外，对于基于二进制文件的接收器，还可以选择使用关键字 $$LASTMODIFIED 添加（源文件的）“上次修改”日期/时间，以及将自定义值作为元数据添加到接收器文件中。

架构和数据类型映射

有关复制活动如何将源数据映射到接收器的信息，请参阅架构和数据类型映射。

在复制过程中添加其他列

除了将数据从源数据存储复制到接收器外，你还可以配置在复制到接收器时添加额外的数据列。 例如：

• 从基于文件的源进行复制时，将相对文件路径存储为字符串类型的附加列，以便跟踪数据的来源文件。
• 将指定的源列复制为另一列。
• 添加一个包含 ADF 表达式的列，用于附加 ADF 系统变量（如管道名称/管道 ID），或者存储来自上游活动输出的其他动态值。
• 添加一个包含静态值的列以满足下游消耗需求。

你可以在复制活动的“源”选项卡中找到以下配置。你还可以像往常一样，在复制活动的架构映射中使用自己定义的列名来映射这些附加列。

若要以编程方式对其进行配置，请在复制活动源中添加 additionalColumns 属性：

示例：

"activities":[
    {
        "name": "CopyWithAdditionalColumns",
        "type": "Copy",
        "inputs": [...],
        "outputs": [...],
        "typeProperties": {
            "source": {
                "type": "<source type>",
                "additionalColumns": [
                    {
                        "name": "filePath",
                        "value": "$$FILEPATH"
                    },
                    {
                        "name": "newColName",
                        "value": "$$COLUMN:SourceColumnA"
                    },
                    {
                        "name": "pipelineName",
                        "value": {
                            "value": "@pipeline().Pipeline",
                            "type": "Expression"
                        }
                    },
                    {
                        "name": "staticValue",
                        "value": "sampleValue"
                    }
                ],
                ...
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

自动创建接收器表

将数据复制到 SQL 数据库/Azure Synapse Analytics 时，如果目标表不存在，复制活动支持基于源数据自动创建它。 它旨在帮助你快速开始加载数据并评估 SQL 数据库/Azure Synapse Analytics。 进行数据引入之后，你可以根据需要检查和调整接收器表架构。

当将数据从任意源复制到以下接收器数据存储时，此功能受支持。 可以在“ADF 创作 UI”–“复制活动接收器”–>“表选项”–>“自动创建表”上，或通过复制活动接收器有效负载中的 属性找到该选项。>tableOption

• Azure SQL 数据库
• Azure SQL 托管实例
• Azure Synapse Analytics
• SQL Server

容错

默认情况下，如果源数据行与接收器数据行不兼容，则复制活动将停止复制数据，并返回失败结果。 要使复制成功，可将复制活动配置为跳过并记录不兼容的行，仅复制兼容的数据。 有关详细信息，请参阅复制活动容错。

数据一致性验证

将数据从源移动到目标存储时，复制活动提供了一个选项，用于执行额外的数据一致性验证，以确保数据不仅成功地从源存储复制到目标存储，而且验证在源存储与目标存储之间保持一致。 在数据移动过程中发现不一致的文件后，你可以中止复制活动，或者通过启用容错设置跳过不一致的文件来继续复制其余的文件。 通过在复制活动中启用会话日志设置，可以获取跳过的文件名称。 有关详细信息，请参阅复制活动中的数据一致性验证。

会话日志

可以记录复制的文件名，这有助于进一步确保数据不仅从源存储成功复制到目标存储，还可以通过查看复制活动会话日志在源和目标存储之间保持一致。 有关详细信息，请参阅复制活动中的会话日志。

相关内容

请参阅以下快速入门、教程和示例：

• 将数据从一个位置复制到同一 Azure Blob 存储帐户中的另一个位置
• 将数据从 Azure Blob 存储复制到 Azure SQL 数据库
• 将数据从 SQL Server 数据库复制到 Azure