数据集成工具：Azure Data Factory：4.数据集成工具：复制活动详解

数据集成工具：AzureDataFactory：4.数据集成工具：复制活动详解1数据集成工具：AzureDataFactory：复制活动详解1.1介绍复制活动1.1.1复制活动概述AzureDataFactory的复制活动是一种用于将数据从一个数据存储快速移动到另一个数据存储的高效工具。它支持多种数据源和接收器，包括但不限于AzureBlob存储、AzureSQL数据库、Azure数据湖存储、AmazonS3、Oracle数据库、SQLServer等。复制活动利用了AzureDataFactory的优化数据移动引擎，能够实现高速、可靠的数据传输。1.1.2复制活动的工作原理复制活动在执行时，会根据源和接收器的类型选择最合适的传输机制。例如，当源和接收器都是基于云的存储时，它会使用直接的云到云传输，避免了不必要的数据下载和上传过程，从而提高了传输效率。对于本地到云的数据传输，复制活动会利用自托管集成运行时（IR）来处理数据移动，确保数据的安全性和传输速度。示例：从AzureBlob存储复制数据到AzureSQL数据库{

"name":"CopyBlobToSQL",

"properties":{

"activities":[

{

"name":"CopyBlobToSQL",

"type":"Copy",

"inputs":[

{

"referenceName":"BlobDataset",

"type":"DatasetReference"

}

],

"outputs":[

{

"referenceName":"SQLDataset",

"type":"DatasetReference"

}

],

"typeProperties":{

"source":{

"type":"BlobSource",

"recursive":true

},

"sink":{

"type":"SqlSink",

"sqlWriterStoredProcedureName":"usp_InsertData"

},

"dataFlow":{

"type":"DataFlow",

"dataFlowName":"DataFlowBlobToSQL"

}

}

}

],

"annotations":[]

}

}在这个示例中，我们定义了一个名为CopyBlobToSQL的管道，它包含一个复制活动。该活动从AzureBlob存储（BlobDataset）读取数据，并将其写入AzureSQL数据库（SQLDataset）。recursive属性设置为true表示复制活动将递归地读取Blob存储中的所有文件。sqlWriterStoredProcedureName属性指定了SQL数据库中用于插入数据的存储过程。1.1.3复制活动的使用场景复制活动适用于以下几种常见的数据集成场景：数据迁移：当需要将数据从旧系统迁移到Azure中的新系统时，复制活动可以快速、高效地完成数据迁移任务。数据备份：定期使用复制活动将数据从生产环境复制到备份存储，可以实现数据的备份和恢复策略。数据同步：在需要保持多个数据存储之间数据一致性的场景下，复制活动可以作为数据同步的工具，确保数据的实时更新。数据处理前的准备：在数据进入复杂的数据处理流程之前，使用复制活动可以将数据从原始存储移动到更易于处理的存储中，如Azure数据湖存储。1.2结论通过上述内容，我们深入了解了AzureDataFactory中复制活动的原理、工作方式以及其在数据集成项目中的应用。复制活动作为AzureDataFactory的核心功能之一，为数据的高效移动提供了强大的支持，是实现数据集成和数据迁移项目的重要工具。2数据集成工具：AzureDataFactory：复制活动详解2.1设置复制活动2.1.1创建复制活动在AzureDataFactory中，复制活动是一种用于将数据从一个数据存储移动到另一个数据存储的高效工具。要创建复制活动，首先需要在AzureDataFactory中创建一个数据工厂，然后在管道中添加复制活动。登录Azure门户，访问AzureDataFactory服务。创建数据工厂，如果尚未创建。打开数据工厂，选择“创建和监控”选项。创建管道，在“活动”工具箱中，拖放“复制数据”活动到设计画布上。配置复制活动，在管道设计画布上，双击复制活动以打开配置窗口。示例代码{

"name":"CopyActivitySample",

"properties":{

"activities":[

{

"name":"CopyFromBlobToSQL",

"type":"Copy",

"typeProperties":{

"source":{

"type":"BlobSource",

"recursive":true

},

"sink":{

"type":"SqlSink",

"sqlWriterStoredProcedureName":"usp_InsertData"

},

"dataset":{

"type":"BlobDataset",

"linkedService":{

"referenceName":"AzureBlobStorage",

"type":"LinkedServiceReference"

},

"typeProperties":{

"folderPath":"inputdata"

}

},

"sink":{

"type":"SqlDataset",

"linkedService":{

"referenceName":"AzureSqlDatabase",

"type":"LinkedServiceReference"

},

"typeProperties":{

"tableName":"dbo.OutputTable"

}

}

}

}

]

}

}2.1.2配置数据源和接收器在复制活动的配置中，需要指定数据源和接收器。数据源是数据的来源，接收器是数据的目标位置。AzureDataFactory支持多种数据存储作为数据源和接收器，包括AzureBlob存储、AzureSQL数据库、AzureCosmosDB等。示例配置在配置窗口中，选择“源”选项卡，然后选择数据源类型，例如“Blob存储”。接着，选择“接收器”选项卡，选择接收器类型，例如“SQL数据库”。2.1.3选择复制数据的类型复制活动支持多种数据类型，包括结构化数据和非结构化数据。在配置复制活动时，需要选择要复制的数据类型。例如，如果数据源是CSV文件，那么在配置中应选择“CSV”作为数据类型。示例选择在配置窗口的“源”选项卡中，选择“CSV”作为数据格式。在“接收器”选项卡中，确保数据类型与目标数据库表匹配。2.1.4设置复制活动的性能和压缩选项为了优化复制活动的性能，可以设置并行度和压缩选项。并行度决定了同时处理的数据文件数量，而压缩选项可以减少传输的数据量，从而提高传输速度。示例设置在配置窗口中，选择“设置”选项卡，可以调整“并行度”和启用“压缩”选项。例如，设置并行度为4，启用GZip压缩。{

"name":"CopyActivityPerformanceSettings",

"properties":{

"activities":[

{

"name":"CopyFromBlobToSQL",

"type":"Copy",

"typeProperties":{

"source":{

"type":"BlobSource",

"recursive":true

},

"sink":{

"type":"SqlSink",

"sqlWriterStoredProcedureName":"usp_InsertData"

},

"copySink":{

"parallelCopies":4,

"compression":{

"type":"GZip",

"level":"Optimal"

}

}

}

}

]

}

}通过以上步骤，可以详细地设置和配置AzureDataFactory中的复制活动，以满足数据集成和传输的需求。3数据集成工具：AzureDataFactory：复制活动的高级功能3.1使用动态内容在AzureDataFactory中，复制活动可以利用动态内容来增强数据迁移的灵活性。动态内容允许您在复制活动的配置中使用表达式，从而根据运行时的条件动态地选择源和目标数据集。这在处理动态变化的文件路径、文件名或数据库表时特别有用。3.1.1示例：动态文件路径复制假设您有一个AzureBlob存储，其中包含多个文件，文件名根据日期动态变化。您希望将这些文件复制到另一个Blob存储中，但目标路径也应根据源文件的日期动态生成。以下是如何使用动态内容实现这一目标的示例：{

"name":"DynamicCopyPipeline",

"properties":{

"activities":[

{

"name":"CopyBlobToBlob",

"type":"Copy",

"typeProperties":{

"source":{

"type":"BlobSource",

"recursive":true

},

"sink":{

"type":"BlobSink"

},

"dataset":{

"type":"BlobDataset",

"linkedService":{

"referenceName":"SourceBlobStorage",

"type":"LinkedServiceReference"

},

"typeProperties":{

"folderPath":{

"value":"@pipeline().parameters.sourceFolderPath",

"type":"Expression"

},

"fileName":{

"value":"@pipeline().parameters.sourceFileName",

"type":"Expression"

}

}

},

"sink":{

"type":"BlobDataset",

"linkedService":{

"referenceName":"DestinationBlobStorage",

"type":"LinkedServiceReference"

},

"typeProperties":{

"folderPath":{

"value":"@concat('destinationFolder/',pipeline().parameters.sourceFileName,'/',pipeline().parameters.sourceFolderPath)",

"type":"Expression"

}

}

}

},

"parameters":{

"sourceFolderPath":{

"type":"string"

},

"sourceFileName":{

"type":"string"

}

}

}

]

}

}在这个例子中，sourceFolderPath和sourceFileName是管道参数，它们在管道运行时动态提供。表达式@pipeline().parameters.sourceFolderPath和@pipeline().parameters.sourceFileName用于动态生成源和目标的文件路径。3.2执行条件复制AzureDataFactory的复制活动可以通过条件表达式来控制数据的复制流程。这意味着您可以根据特定条件决定是否执行复制，或者复制哪些数据。这在处理数据时提供了额外的控制层，确保只有满足特定条件的数据才会被迁移。3.2.1示例：基于文件大小的条件复制假设您有一个数据湖，其中包含大量文件，但您只希望复制大小超过1GB的文件。以下是如何使用条件表达式来实现这一需求的示例：{

"name":"ConditionalCopyPipeline",

"properties":{

"activities":[

{

"name":"LookupFileSize",

"type":"Lookup",

"typeProperties":{

"source":{

"type":"BlobSource",

"recursive":true,

"enablePartitionDiscovery":true

},

"dataset":{

"type":"BlobDataset",

"linkedService":{

"referenceName":"SourceBlobStorage",

"type":"LinkedServiceReference"

},

"typeProperties":{

"folderPath":"sourceFolder"

}

}

}

},

{

"name":"CopyLargeFiles",

"type":"Copy",

"typeProperties":{

"source":{

"type":"BlobSource"

},

"sink":{

"type":"BlobSink"

},

"dataset":{

"type":"BlobDataset",

"linkedService":{

"referenceName":"DestinationBlobStorage",

"type":"LinkedServiceReference"

},

"typeProperties":{

"folderPath":"destinationFolder"

}

},

"conditions":[

{

"expression":{

"value":"@activity('LookupFileSize').output.firstRow.file_size>1073741824",

"type":"Expression"

}

}

]

}

}

]

}

}在这个例子中，首先使用Lookup活动来检索源文件的大小。然后，CopyLargeFiles活动通过条件@activity('LookupFileSize').output.firstRow.file_size>1073741824来判断文件大小是否超过1GB，只有满足条件的文件才会被复制。3.3复制活动的错误处理和重试策略在数据复制过程中，可能会遇到各种错误，如网络问题、权限问题或数据格式不匹配等。AzureDataFactory提供了错误处理和重试策略，以确保数据复制的可靠性和数据的完整性。3.3.1错误处理AzureDataFactory允许您在复制活动的配置中指定错误处理策略。例如，您可以选择在遇到错误时停止整个管道，或者跳过错误的文件并继续处理其他文件。3.3.2重试策略重试策略允许您在遇到暂时性错误时自动重试复制活动。这可以提高管道的健壮性，减少因短暂网络中断或资源不可用导致的失败。3.3.3示例：配置重试策略和错误处理以下是一个配置了重试策略和错误处理的复制活动示例：{

"name":"RetryAndErrorHandlingPipeline",

"properties":{

"activities":[

{

"name":"CopyBlobToBlob",

"type":"Copy",

"typeProperties":{

"source":{

"type":"BlobSource"

},

"sink":{

"type":"BlobSink"

},

"dataset":{

"type":"BlobDataset",

"linkedService":{

"referenceName":"SourceBlobStorage",

"type":"LinkedServiceReference"

},

"typeProperties":{

"folderPath":"sourceFolder"

}

},

"sink":{

"type":"BlobDataset",

"linkedService":{

"referenceName":"DestinationBlobStorage",

"type":"LinkedServiceReference"

},

"typeProperties":{

"folderPath":"destinationFolder"

}

},

"retryPolicy":{

"count":3,

"intervalInSeconds":30

},

"onError":{

"action":"Continue",

"skipExpression":{

"value":"@activity('CopyBlobToBlob').output.errors[0].errorCode=='404'",

"type":"Expression"

}

}

}

}

]

}

}在这个例子中，retryPolicy配置了在遇到错误时自动重试3次，每次重试间隔30秒。onError配置则指定了在遇到特定错误（如文件不存在错误）时，管道应继续执行，而不是停止。通过这些高级功能，AzureDataFactory的复制活动可以更智能、更灵活地处理复杂的数据迁移场景，确保数据的准确性和管道的稳定性。4监控和优化AzureDataFactory中的复制活动4.1监控复制活动的执行4.1.1监控工具与指标AzureDataFactory提供了多种工具来监控复制活动的执行情况，包括Azure门户、DataFactory监控工具、日志分析以及AzureMonitor。这些工具可以帮助你追踪活动的状态、执行时间、数据传输速率等关键指标。示例：使用AzureMonitor监控复制活动#使用AzureCLI查询复制活动的监控数据

azmonitormetricslist\

--resource{your-data-factory-resource-id}\

--metric"DataRead""DataWritten""CopyDuration"\

--intervalPT1H\

--aggregationtotal上述代码示例展示了如何使用AzureCLI来查询复制活动的监控数据，包括读取的数据量(DataRead)、写入的数据量(DataWritten)以及复制活动的持续时间(CopyDuration)。通过设置不同的聚合方式和时间间隔，可以获取更详细的监控信息。4.1.2诊断复制活动问题当复制活动出现故障时，可以利用DataFactory的日志和诊断设置来定位问题。例如，检查活动运行日志中的错误消息，或者使用AzureMonitor的日志查询功能来分析更深层次的执行细节。示例：分析复制活动日志在Azure门户中，导航至你的DataFactory实例，选择“监控”>“活动运行”，然后筛选出复制活动。点击具体的活动运行记录，可以查看详细的执行日志，包括源和接收器的性能统计、错误信息等。4.2分析复制活动的性能4.2.1性能指标为了优化复制活动，需要关注以下性能指标：-数据传输速率：衡量数据从源到接收器的传输速度。-CPU利用率：了解执行活动的计算资源使用情况。-I/O等待时间：分析数据读写操作的等待时间，以判断是否存在I/O瓶颈。示例：分析数据传输速率假设你正在监控一个从AzureBlob存储复制数据到AzureSQL数据库的复制活动。在AzureMonitor中，你可以设置一个图表来显示“DataRead”和“DataWritten”的指标，以秒为单位，从而计算出平均数据传输速率。4.3优化复制活动的策略4.3.1并行复制AzureDataFactory支持并行复制数据，通过增加并行度可以显著提高数据传输速率。并行度的设置取决于源和接收器的性能，以及网络带宽。示例：设置并行复制在复制活动的设置中，可以调整“并行复制”选项。例如，如果源数据存储在AzureBlob，而接收器是AzureSQL数据库，可以尝试将并行度设置为4或更高，以优化数据传输。{

"name":"CopyActivity",

"type":"Copy",

"typeProperties":{

"source":{

"type":"BlobSource",

"recursive":true

},

"sink":{

"type":"SqlSink",

"sqlWriterStoredProcedureName":"usp_InsertData"

},

"parallelCopies":4

}

}上述JSON配置展示了如何在复制活动的定义中设置并行复制的参数。parallelCopies的值设置为4，意味着复制活动将尝试同时使用4个数据流进行数据传输。4.3.2选择合适的接收器类型不同的接收器类型对数据的处理方式不同，选择最适合你数据特性的接收器类型可以提高复制效率。示例：使用SQL接收器如果你正在从一个结构化数据源复制数据到AzureSQL数据库，使用SQL接收器可以利用SQLServer的批量插入功能，从而提高数据写入速度。{

"name":"CopyActivity",

"type":"Copy",

"typeProperties":{

"sink":{

"type":"SqlSink",

"sqlWriterStoredProcedureName":"usp_BulkInsert"

}

}

}在这个例子中，sqlWriterStoredProcedureName被设置为usp_BulkInsert，这意味着复制活动将调用一个存储过程来执行批量插入操作，而不是逐行插入，从而提高数据写入的效率。4.3.3利用压缩在复制大量数据时，利用压缩可以减少数据传输的时间和成本。AzureDataFactory支持多种压缩格式，如GZip、Deflate等。示例：设置压缩在复制活动的源设置中，可以指定压缩格式和压缩级别。例如，从AzureBlob存储复制压缩的CSV文件到AzureSQL数据库。{

"name":"CopyActivity",

"type":"Copy",

"typeProperties":{

"source":{

"type":"BlobSource",

"compression":{

"type":"GZip",

"level":"Optimal"

}

}

}

}上述JSON配置展示了如何在复制活动的源设置中启用GZip压缩，并设置压缩级别为“Optimal”。这将自动压缩源数据，从而减少数据传输的大小，提高复制效率。4.3.4调整数据流根据源和接收器的性能，调整数据流的大小和类型可以进一步优化复制活动。例如，使用更大的数据流可以提高数据传输速率，但可能会增加计算资源的使用。示例：调整数据流大小在复制活动的设置中，可以调整“数据流大小”选项。如果源数据存储在AzureBlob，而接收器是AzureSQL数据库，可以尝试将数据流大小设置为“Large”，以优化数据传输。{

"name":"CopyActivity",

"type":"Copy",

"typeProperties":{

"dataFlow":{

"type":"Large"

}

}

}请注意，AzureDataFactory的数据流大小调整是一个高级特性，需要根据具体的数据源和接收器类型来决定最佳设置。在上述JSON配置中，dataFlow的类型被设置为“Large”，这将尝试使用更大的数据流进行数据传输，但具体的效果需要通过监控和测试来验证。4.3.5利用缓存在复制活动之间，利用缓存可以减少重复的数据读取操作，从而提高整体的复制效率。AzureDataFactory支持在活动之间共享数据集，从而实现缓存的效果。示例：利用缓存假设你有多个复制活动，都需要从同一个AzureBlob存储读取数据。在第一个复制活动完成后，可以将数据集设置为“缓存”，这样后续的复制活动可以直接从缓存中读取数据，而不需要再次访问Blob存储。{

"name":"CachedDataset",

"properties":{

"type":"AzureBlob",

"linkedServiceName":{

"referenceName":"AzureStorageLinkedService",

"type":"LinkedServiceReference"

},

"typeProperties":{

"fileName":"data.csv",

"folderPath":"input/",

"format":{

"type":"TextFormat",

"columnDelimiter":","

}

},

"cacheMode":"IfSourceModified"

}

}在上述JSON配置中，cacheMode被设置为“IfSourceModified”，这意味着只有当源数据发生变化时，才会重新读取数据。否则，后续的复制活动将直接从缓存中读取数据，从而提高效率。4.3.6总结通过监控复制活动的执行、分析性能指标、并采取上述优化策略，可以显著提高AzureDataFactory中复制活动的效率和可靠性。务必根据你的具体需求和资源限制，调整并行度、接收器类型、压缩设置以及数据流大小，以达到最佳的复制效果。同时，利用缓存可以减少重复的数据读取操作，进一步提高整体的复制效率。5数据集成工具：AzureDataFactory：复制活动详解5.1复制活动的常见问题与解决方案5.1.1问题诊断问题1:数据复制失败，源数据与目标数据类型不匹配场景描述:在使用AzureDataFactory的复制活动时，如果源数据类型与目标数据类型不匹配，可能会导致数据复制失败。解决方案:1.检查数据类型:确保源数据表和目标数据表的列数据类型一致。2.使用转换:如果数据类型不一致，可以使用转换活动（如映射数据流）来转换数据类型，然后再进行复制。示例代码:{

"name":"MappingDataFlow",

"properties":{

"type":"MappingDataFlow",

"typeProperties":{

"sources":[

{

"name":"sourceDataset",

"properties":{

"type":"AzureSqlSource",

"sqlReaderQuery":"SELECT*FROMsourceTable",

"typeProperties":{

"type":"AzureSqlSource",

"sqlReaderQuery":"SELECT*FROMsourceTable",

"sqlReaderStoredProcedureName":"",

"storedProcedureParameters":{},

"partitioningMode":"none",

"additionalColumns":""

}

}

}

],

"sinks":[

{

"name":"sinkDataset",

"properties":{

"type":"AzureSqlSink",

"typeProperties":{

"type":"AzureSqlSink",

"preCopyScript":"",

"postCopyScript":"",

"sqlWriterStoredProcedureName":"",

"storedProcedureParameters":{},

"tableReference":"targetTable",

"additionalColumns":""

}

}

}

],

"transformations":[

{

"name":"typeConversion",

"properties":{

"type":"DerivedColumn",

"inputs":[

"sourceDataset"

],

"outputs":[

"convertedData"

],

"transformation":{

"type":"DerivedColumn",

"inputs":[

"sourceDataset"

],

"outputs":[

{

"name":"convertedData",

"properties":{

"structure":[

{

"name":"column1",

"type":"String"

},

{

"name":"column2",

"type":"Int32"

}

]

}

}

],

"expressions":[

{

"name":"column1",

"expression":"@toString($$Text(1))"

},

{

"name":"column2",

"expression":"@toInt($$Text(2))"

}

]

}

}

}

]

}

}

}描述:此示例展示了如何使用映射数据流中的DerivedColumn转换来将源数据中的列数据类型从Text转换为String和Int32，以匹配目标数据表的列数据类型。问题2:复制活动性能低下场景描述:当复制大量数据时，复制活动的性能可能成为瓶颈，导致数据加载时间过长。解决方案:1.增加并行度:在复制活动的设置中，增加并行度可以提高数据复制的速度。2.优化数据源:确保数据源的读取性能，例如使用索引、分区等技术。3.使用压缩:对数据进行压缩可以减少传输的数据量，从而提高复制速度。示例代码:{

"name":"CopyActivity",

"properties":{

"type":"Copy",

"typeProperties":{

"source":{

"type":"AzureSqlSource",

"sqlReaderQuery":"SELECT*FROMsourceTable",

"typeProperties":{

"type":"AzureSqlSource",

"sqlReaderQuery":"SELECT*FROMsourceTable",

"sqlReaderStoredProcedureName":"",

"storedProcedureParameters":{},

"partitioningMode":"none",

"additionalColumns":""

}

},

"sink":{

"type":"AzureSqlSink",

"typeProperties":{

"type":"AzureSqlSink",

"preCopyScript":"",

"postCopyScript":"",

"sqlWriterStoredProcedureName":"",

"storedProcedureParameters":{},

"tableReference":"targetTable",

"additionalColumns":""

}

},

"datasetParameters":{},

"copyBehavior":"None",

"copyConcurrency":10,//增加并行度

"sinkRetryWrite":0,

"sourceRetryRead":0,

"sinkWriteBatchSize":0,

"sinkWriteBatchTimeout":"00:00:00",

"enableStaging":false,

"staging":{},

"enableSkipIncompatibleRow":false,

"redirectIncompatibleRow":"",

"enableStaleDataDetection":false,

"staleDataDetectionPolicy":{},

"enableDataConsistencyCheck":false,

"dataConsistencyCheckPolicy":{},

"enablePartitionDiscovery":false,

"partitionSettings":{},

"enableSchemaDriftManagement":false,

"schemaDriftPolicy":{},

"enableSkipIncompatibleRowAndReportResult":false,

"redirectIncompatibleRowAndReportResult":"",

"enableStagingForLargeData":false,

"stagingForLargeData":{},

"enableStagingForSmallData":false,

"stagingForSmallData":{},

"enableStagingForAllData":false,

"stagingForAllData":{},

"enableStagingForLargeDataAndSmallData":false,

"stagingForLargeDataAndSmallData":{},

"enableStagingForAllDataAndSmallData":false,

"stagingForAllDataAndSmallData":{},

"enableStagingForAllDataAndLargeData":false,

"stagingForAllDataAndLargeData":{},

"enableStagingForAllDataAndLargeDataAndSmallData":false,

"stagingForAllDataAndLargeDataAndSmallData":{},

"enableStagingForLargeDataAndSmallDataAndAllData":false,

"stagingForLargeDataAndSmallDataAndAllData":{},

"enableStagingForSmallDataAndAllData":false,

"stagingForSmallData