数据集成工具：Azure Data Factory：连接到各种数据源

数据集成工具：AzureDataFactory：连接到各种数据源1数据源连接概述1.1AzureDataFactory支持的数据源AzureDataFactory(ADF)是一个云端数据整合服务，用于创建和调度数据驱动的工作流。它支持多种数据源，包括但不限于：SQLServerAzureSQLDatabaseAzureCosmosDBAzureBlobStorageAzureDataLakeStorageAmazonS3OracleDatabaseMongoDBAzureSynapseAnalyticsAzureTableStorageAzureCosmosDBSQLAPIAzureCosmosDBMongoDBAPIAzureSQLManagedInstancePostgreSQLMySQLTeradataSAPHANASAPBWSalesforceGoogleBigQueryHadoopHDFSHadoopHiveHadoopPigHadoopMapReduceHadoopStreamingHadoopDistCpHadoopImpalaHadoopSparkHadoopHiveonHDInsightHadoopPigonHDInsightHadoopMapReduceonHDInsightHadoopStreamingonHDInsightHadoopDistCponHDInsightHadoopImpalaonHDInsightHadoopSparkonHDInsightAzureEventHubsAzureQueueStorageAzureSearchAzureEventHubsCaptureAzureStreamAnalyticsAzureAnalysisServicesAzureDatabricksAzureDatabaseforMySQLAzureDatabaseforPostgreSQLAzureDatabaseforMariaDBAzureDatabaseforSQLServerAzureDatabaseforPostgreSQL-FlexibleServerAzureDatabaseforMySQL-FlexibleServerAzureDatabaseforMariaDB-FlexibleServerAzureDatabaseforSQLServer-ManagedInstanceAzureDatabaseforPostgreSQL-SingleServerAzureDatabaseforMySQL-SingleServerAzureDatabaseforMariaDB-SingleServerAzureDatabaseforSQLServer-SingleServerAzureDatabaseforPostgreSQL-HyperscaleAzureDatabaseforMySQL-HyperscaleAzureDatabaseforMariaDB-HyperscaleAzureDatabaseforSQLServer-HyperscaleAzureDatabaseforPostgreSQL-ServerlessAzureDatabaseforMySQL-ServerlessAzureDatabaseforMariaDB-ServerlessAzureDatabaseforSQLServer-ServerlessAzureDatabaseforPostgreSQL-ManagedInstanceAzureDatabaseforMySQL-ManagedInstanceAzureDatabaseforMariaDB-ManagedInstanceAzureDatabaseforSQLServer-ManagedInstanceAzureDatabaseforPostgreSQL-DedicatedAzureDatabaseforMySQL-DedicatedAzureDatabaseforMariaDB-DedicatedAzureDatabaseforSQLServer-DedicatedAzureDatabaseforPostgreSQL-GeneralPurposeAzureDatabaseforMySQL-GeneralPurposeAzureDatabaseforMariaDB-GeneralPurposeAzureDatabaseforSQLServer-GeneralPurposeAzureDatabaseforPostgreSQL-MemoryOptimizedAzureDatabaseforMySQL-MemoryOptimizedAzureDatabaseforMariaDB-MemoryOptimizedAzureDatabaseforSQLServer-MemoryOptimized这些数据源可以通过ADF的连接器进行集成，实现数据的复制、转换和加载。1.2连接器的类型和功能1.2.1SQLServer连接器原理SQLServer连接器用于从本地或Azure虚拟机上的SQLServer读取数据，或向其写入数据。它使用SQLServer驱动程序通过ODBC或JDBC连接到数据库。示例{

"name":"SqlServerConnection",

"properties":{

"type":"SqlServer",

"typeProperties":{

"server":"<your_server_name>",

"database":"<your_database_name>",

"authenticationType":"Basic",

"username":"<your_username>",

"password":{

"type":"SecureString",

"value":"<your_password>"

}

},

"connectVia":{

"referenceName":"<your_integration_runtime>",

"type":"IntegrationRuntimeReference"

}

}

}server:SQLServer的名称或IP地址。database:要连接的数据库名称。authenticationType:认证类型，可以是基本认证或Windows认证。username和password:数据库的登录凭证。connectVia:指定的集成运行时，用于处理本地或虚拟机上的数据源。1.2.2AzureSQLDatabase连接器原理AzureSQLDatabase连接器用于连接到AzureSQLDatabase，它使用SQLServer驱动程序通过ODBC或JDBC进行连接。示例{

"name":"AzureSqlDatabaseConnection",

"properties":{

"type":"AzureSqlDatabase",

"typeProperties":{

"server":"<your_server_name>.",

"database":"<your_database_name>",

"authentication":{

"type":"ManagedIdentity",

"managedIdentityResourceId":"<your_managed_identity_resource_id>"

}

}

}

}server:AzureSQLDatabase的服务器名称。database:数据库名称。authentication:使用托管身份进行认证，无需存储用户名和密码。1.2.3AzureCosmosDB连接器原理AzureCosmosDB连接器用于连接到AzureCosmosDB，支持SQLAPI和MongoDBAPI。它使用CosmosDB的SDK进行连接。示例{

"name":"CosmosDbConnection",

"properties":{

"type":"CosmosDb",

"typeProperties":{

"connectionString":{

"type":"SecureString",

"value":"AccountEndpoint=<your_account_endpoint>;AccountKey=<your_account_key>;Database=<your_database_name>"

}

}

}

}connectionString:包含账户端点、账户密钥和数据库名称的连接字符串。1.2.4使用连接器进行数据复制原理使用ADF的复制活动，可以从一个数据源读取数据并将其写入另一个数据源。这需要在源和接收器之间定义连接器。示例{

"name":"CopyFromSQLToCosmosDB",

"properties":{

"activities":[

{

"name":"CopySQLtoCosmosDB",

"type":"Copy",

"inputs":[

{

"referenceName":"SqlServerConnection",

"type":"DatasetReference"

}

],

"outputs":[

{

"referenceName":"CosmosDbConnection",

"type":"DatasetReference"

}

],

"typeProperties":{

"source":{

"type":"SqlSource",

"sqlReaderQuery":"SELECT*FROM<your_table_name>"

},

"sink":{

"type":"CosmosDbSink",

"writeBehavior":"insert"

}

}

}

]

}

}source:定义SQLServer数据源的读取方式。sink:定义AzureCosmosDB数据接收器的写入方式。1.2.5总结通过上述连接器和示例，可以理解如何在AzureDataFactory中连接和操作各种数据源。这为数据工程师提供了强大的工具，以实现数据的高效集成和处理。2数据集成工具：AzureDataFactory连接到SQLServer2.1配置SQLServer连接在AzureDataFactory中，连接到SQLServer数据源需要创建一个链接服务。链接服务定义了数据工厂与数据源之间的连接属性。以下是创建SQLServer链接服务的步骤：打开AzureDataFactory：登录Azure门户，选择你的数据工厂，然后在“数据工厂”菜单中选择“数据集成”>“数据工厂”>“编辑”。创建链接服务：在左侧菜单中，选择“链接服务”，然后点击“新建链接服务”。选择数据源类型：在“新建链接服务”窗口中，选择“SQLServer”。输入连接属性：服务器名称：输入SQLServer的服务器名称。数据库名称：输入SQLServer的数据库名称。身份验证类型：选择身份验证类型，如“基本”或“Windows”。用户名：输入用于连接SQLServer的用户名。密码：输入用于连接SQLServer的密码。连接字符串：可选，直接输入连接字符串。测试连接：点击“测试连接”按钮，确保连接成功。保存链接服务：输入链接服务的名称，然后点击“保存”。2.1.1示例代码：创建SQLServer链接服务{

"name":"SqlServer1",

"properties":{

"type":"SqlServer",

"typeProperties":{

"server":"",

"database":"yourdatabase",

"authenticationType":"Basic",

"username":"yourusername",

"password":"yourpassword"

},

"connectVia":{

"referenceName":"IntegrationRuntime1",

"type":"IntegrationRuntimeReference"

}

}

}2.2使用SQLServer数据源的活动一旦创建了SQLServer的链接服务，你就可以在数据工厂的管道中使用它来执行各种活动，如复制数据、查询数据等。2.2.1复制数据活动示例假设你有一个SQLServer数据库，你想要将其中的数据复制到AzureBlob存储中。创建管道：在“编辑”界面中，选择“管道”，然后点击“新建”。添加复制数据活动：从“活动”工具箱中，拖拽“复制数据”活动到画布上。配置复制数据活动：源数据集：选择或创建一个SQLServer数据集，与你的SQLServer链接服务关联。接收数据集：选择或创建一个AzureBlob存储数据集。源数据源：选择你之前创建的SQLServer链接服务。接收数据源：选择你的AzureBlob存储链接服务。设置查询：在“复制数据”活动的“源”设置中，你可以设置一个SQL查询来指定要复制的数据。2.2.2示例代码：复制数据活动{

"name":"CopyFromSQLServerToBlob",

"properties":{

"activities":[

{

"name":"CopySQLServerToBlob",

"type":"Copy",

"inputs":[

{

"referenceName":"SQLServerDataset",

"type":"DatasetReference"

}

],

"outputs":[

{

"referenceName":"BlobDataset",

"type":"DatasetReference"

}

],

"typeProperties":{

"source":{

"type":"SqlSource",

"sqlReaderQuery":"SELECT*FROMyourtable"

},

"sink":{

"type":"BlobSink"

}

}

}

]

}

}2.2.3解释在上述代码中，我们定义了一个管道，其中包含一个复制数据活动。该活动从SQLServer数据源读取数据，并将其复制到AzureBlob存储中。我们通过设置sqlReaderQuery来指定要从SQLServer中复制的数据表。通过这些步骤和示例，你可以有效地在AzureDataFactory中配置SQLServer的连接，并使用它来执行数据集成任务。这为从SQLServer数据库中提取、转换和加载数据提供了强大的工具。3连接到AzureSQL数据库3.1创建AzureSQL数据库连接在使用AzureDataFactory进行数据集成时，连接到AzureSQL数据库是关键步骤之一。以下是如何在AzureDataFactory中创建AzureSQL数据库连接的详细步骤：登录Azure门户：首先，登录到Azure门户(/)。访问DataFactory：在Azure门户的左侧菜单中，选择“所有服务”，然后搜索并选择“DataFactory”。找到你的DataFactory实例并点击进入。创建连接：在DataFactory实例的左侧菜单中，选择“连接”选项。点击“新建连接”按钮。选择数据源类型：在“新建连接”窗口中，选择“AzureSQL数据库”作为数据源类型。输入连接详细信息：在“连接详细信息”部分，输入你的AzureSQL数据库的详细信息，包括服务器名称、数据库名称、用户名和密码。确保这些信息是正确的，以便成功连接。测试连接：输入完详细信息后，点击“测试连接”按钮，以验证连接设置是否正确。保存连接：如果测试成功，点击“创建”按钮保存连接。现在，你可以在DataFactory的管道中使用这个连接了。3.2执行AzureSQL数据库查询一旦创建了AzureSQL数据库的连接，你就可以在AzureDataFactory的管道中执行查询，以读取、写入或更新数据。以下是如何在管道中使用AzureSQL数据库连接执行查询的步骤：创建管道：在DataFactory实例中，选择“创建和监控”选项，然后点击“创建管道”。添加源数据集：在管道设计画布中，点击“源”图标，选择“AzureSQL数据库”。在“新建数据集”窗口中，选择你之前创建的AzureSQL数据库连接。配置查询：在数据集配置窗口中，选择“表”或“SQL查询”作为数据源类型。如果选择“SQL查询”，你可以在“SQL查询”字段中输入自定义的SQL查询语句。例如，假设你有一个名为Sales的表，你可以使用以下SQL查询来读取数据：SELECT*FROMSalesWHERESaleDate>='2023-01-01'添加接收器：在管道设计画布中，添加一个接收器，例如“AzureBlob存储”或“AzureSQL数据库”，以将数据写入或更新到目标位置。创建数据流：将源数据集拖放到接收器上，创建数据流。在数据流中，你可以进行数据转换，如筛选、映射字段等。配置接收器：在接收器配置窗口中，选择目标位置和写入模式。如果目标是另一个AzureSQL数据库，你可以选择“表”或“SQL命令”来写入数据。例如，将数据写入名为SalesSummary的表中，你可以使用以下SQL命令：INSERTINTOSalesSummary(TotalSales,SaleDate)

SELECTSUM(SaleAmount),SaleDateFROMSalesGROUPBYSaleDate保存并发布管道：完成管道设计后，点击“保存”按钮，然后在“创建和监控”页面中，点击“发布所有内容”以发布管道。触发管道执行：在“创建和监控”页面中，选择“触发器”，然后创建一个新的触发器来自动执行管道，或手动触发管道执行。通过以上步骤，你可以在AzureDataFactory中创建AzureSQL数据库的连接，并使用这些连接来执行复杂的SQL查询，从而实现数据的读取、写入和更新。这为数据集成和处理提供了强大的工具，使你能够轻松地在不同的数据源之间移动和转换数据。4连接到AzureCosmosDB4.1理解CosmosDB连接器AzureDataFactory(ADF)提供了CosmosDB连接器，用于在ADF管道中读取和写入AzureCosmosDB数据。CosmosDB是一个全球分布式、多模型数据库服务，为应用程序提供低延迟、高吞吐量和高可用性。ADF的CosmosDB连接器支持SQLAPI和MongoDBAPI，允许你使用ADF的复制活动或查询活动来处理数据。4.1.1特性低延迟访问：CosmosDB连接器利用CosmosDB的高性能，提供毫秒级的延迟。数据复制：可以将数据从CosmosDB复制到其他数据存储，或从其他数据存储复制到CosmosDB。数据查询：使用SQL查询或MongoDB命令查询CosmosDB数据。4.1.2支持的活动复制活动：从CosmosDB读取数据并写入到另一个数据存储。查询活动：执行SQL查询或MongoDB命令，获取结果并进行后续处理。4.2配置CosmosDB连接在ADF中配置CosmosDB连接涉及以下步骤：创建连接：在ADF的“连接”区域，选择“新建连接”，然后选择“AzureCosmosDB”。输入连接详细信息：提供CosmosDB的账户名、密钥、数据库名和容器名。选择API类型：根据你的CosmosDB实例选择SQLAPI或MongoDBAPI。测试连接：确保连接信息正确无误，可以成功连接到CosmosDB。4.2.1示例代码{

"name":"CosmosDBConnection",

"properties":{

"type":"CosmosDb",

"typeProperties":{

"accountEndpoint":":443/",

"accountKey":"yourcosmosdbaccountkey==",

"database":"yourdatabase",

"container":"yourcontainer",

"preferredRegions":[

"EastUS",

"WestEurope"

]

},

"connectVia":{

"referenceName":"yourintegrationruntime",

"type":"IntegrationRuntimeReference"

}

}

}4.3使用CosmosDB进行数据复制使用ADF的复制活动，你可以将数据从CosmosDB复制到另一个数据存储，或反之。以下是一个使用复制活动从CosmosDB读取数据并写入到AzureBlob存储的示例。4.3.1示例代码{

"name":"CopyFromCosmosDBToBlob",

"properties":{

"activities":[

{

"name":"CopyCosmosDBToBlob",

"type":"Copy",

"inputs":[

{

"referenceName":"CosmosDBSource",

"type":"DatasetReference"

}

],

"outputs":[

{

"referenceName":"AzureBlobSink",

"type":"DatasetReference"

}

],

"typeProperties":{

"source":{

"type":"CosmosDbSource",

"query":"SELECT*FROMcWHEREc.id='123'"

},

"sink":{

"type":"BlobSink",

"writeBatchSize":0,

"writeBatchTimeout":"00:00:00"

}

}

}

],

"annotations":[]

}

}4.3.2解释活动名称：CopyFromCosmosDBToBlob，定义了管道的主要功能。活动类型：Copy，表示这是一个复制活动。输入数据集：CosmosDBSource，从CosmosDB读取数据。输出数据集：AzureBlobSink，将数据写入AzureBlob存储。查询：SELECT*FROMcWHEREc.id='123'，从CosmosDB中选择特定的数据行。写入设置：writeBatchSize和writeBatchTimeout控制数据写入Blob存储的批量大小和超时时间。通过以上步骤和示例，你可以有效地在AzureDataFactory中配置和使用CosmosDB连接器，实现数据的高效复制和处理。5高级连接策略5.1使用数据管理网关连接本地数据源在AzureDataFactory中，连接本地数据源（如SQLServer）需要通过Azure数据管理网关(DataManagementGateway)。数据管理网关是一个可以在本地网络中安装的软件组件，它充当Azure云服务与本地数据源之间的桥梁，允许数据工厂安全地访问本地数据。5.1.1安装数据管理网关下载网关软件：从Azure门户中下载数据管理网关安装程序。安装网关：在本地网络中的一台计算机上安装网关软件。注册网关：在安装过程中，使用Azure订阅中的资源组和数据工厂的名称来注册网关。5.1.2配置网关网关组：在AzureDataFactory中创建一个网关组，将网关添加到该组中。数据源连接：创建数据源连接时，选择“自托管集成运行时”并关联到相应的网关组。5.1.3示例：使用数据管理网关从本地SQLServer复制数据到AzureSQLDatabase{

"name":"OnPremToAzureSql",

"properties":{

"activities":[

{

"name":"CopyFromOnPremSQLToAzureSQL",

"type":"Copy",

"linkedServiceName":{

"referenceName":"OnPremSQLServer",

"type":"LinkedServiceReference"

},

"inputs":[

{

"referenceName":"OnPremSQLTable",

"type":"DatasetReference"

}

],

"outputs":[

{

"referenceName":"AzureSQLTable",

"type":"DatasetReference"

}

],

"typeProperties":{

"source":{

"type":"SqlSource",

"sqlReaderQuery":"SELECT*FROMMyTable"

},

"sink":{

"type":"SqlSink",

"sqlWriterTableType":"MyTableType"

},

"dataIntegrationRuntime":{

"type":"SelfHostedIntegrationRuntime",

"name":"MyGateway"

}

}

}

]

}

}在上述示例中，OnPremSQLServer是与本地SQLServer连接的链接服务，OnPremSQLTable和AzureSQLTable分别是本地和AzureSQL数据库中的数据集。MyGateway是注册的数据管理网关的名称。5.2配置自托管集成运行时自托管集成运行时(Self-hostedIntegrationRuntime)是AzureDataFactory中用于处理本地数据源或私有网络中数据源的数据移动和数据集成任务的组件。它可以在本地网络中或虚拟网络中运行，提供与Azure云服务之间的数据传输。5.2.1创建自托管集成运行时在AzureDataFactory中创建：通过Azure门户，在数据工厂中创建一个新的自托管集成运行时。下载安装程序：下载自托管集成运行时的安装程序，并在本地网络或虚拟网络中的计算机上安装。5.2.2配置自托管集成运行时安装位置：选择一个具有稳定网络连接的计算机来安装自托管集成运行时。注册运行时：安装后，运行时需要注册到AzureDataFactory中，以便数据工厂可以使用它。5.2.3示例：使用自托管集成运行时从本地文件系统复制数据到AzureBlob存储{

"name":"OnPremFileSystemToAzureBlob",

"properties":{

"activities":[

{

"name":"CopyFromOnPremFileSystemToBlob",

"type":"Copy",

"linkedServiceName":{

"referenceName":"OnPremFileSystem",

"type":"LinkedServiceReference"

},

"inputs":[

{

"referenceName":"OnPremFile",

"type":"DatasetReference"

}

],

"outputs":[

{

"referenceName":"AzureBlob",

"type":"DatasetReference"

}

],

"typeProperties":{

"source":{

"type":"FileSource",

"recursive":true

},

"sink":{

"type":"BlobSink"

},

"dataIntegrationRuntime":{

"type":"SelfHostedIntegrationRuntime",

"name":"MySelfHostedIR"

}

}

}

]

}

}在本例中，OnPremFileSystem是与本地文件系统连接的链接服务，OnPremFile和AzureBlob分别是本地文件和AzureBlob存储中的数据集。MySelfHostedIR是自托管集成运行时的名称。5.3优化数据集成性能数据集成性能的优化是确保数据移动和处理任务高效完成的关键。AzureDataFactory提供了多种策略来优化数据集成性能，包括但不限于：并行处理：通过增加并行度，可以同时处理多个数据流，从而提高数据处理速度。数据压缩：在数据传输过程中使用压缩可以减少数据传输量，从而提高传输速度。使用分区：对于大型数据集，使用分区可以提高数据加载和查询的效率。选择合适的集成运行时：根据数据源的位置和网络状况，选择最合适的集成运行时。5.3.1示例：优化从AzureSQLDatabase到AzureCosmosDB的数据复制性能{

"name":"OptimizedCopyFromAzureSQLToCosmosDB",

"properties":{

"activities":[

{

"name":"CopyFromAzureSQLToCosmosDB",

"type":"Copy",

"linkedServiceName":{

"referenceName":"AzureSQLDatabase",

"type":"LinkedServiceReference"

},

"inputs":[

{

"referenceName":"AzureSQLTable",

"type":"DatasetReference"

}

],

"outputs":[

{

"referenceName":"AzureCosmosDB",

"type":"DatasetReference"

}

],

"typeProperties":{

"source":{

"type":"SqlSource",

"sqlReaderQuery":"SELECT*FROMMyTable"

},

"sink":{

"type":"CosmosDbSink",

"writeBatchSize":1000,

"writeBatchTimeout":"02:00:00"

},

"dataIntegrationRuntime":{

"type":"ManagedIntegrationRuntime",

"name":"Azure"

},

"parallelCopies":5

}

}

]

}

}在本例中，AzureSQLDatabase和AzureCosmosDB分别是与AzureSQLDatabase和AzureCosmosDB连接的链接服务。AzureSQLTable和AzureCosmosDB是数据集。writeBatchSize和writeBatchTimeout用于控制CosmosDBSink的写入批次大小和超时时间，parallelCopies用于设置并行复制的数量。5.3.2总结通过使用数据管理网关和自托管集成运行时，AzureDataFactory能够安全高效地连接和处理本地数据源和私有网络中的数据源。同时，通过优化数据集成策略，可以显著提高数据处理和移动的性能。这些策略包括并行处理、数据压缩、使用分区以及选择合适的集成运行时。正确配置和优化这些组件，对于构建高效的数据集成解决方案至关重要。6连接器的管理和监控6.1管理连接器的生命周期在AzureDataFactory中，连接器是用于与不同数据源建立连接的关键组件。管理连接器的生命周期包括创建、更新、测试和删除连接器。下面我们将通过具体的步骤和代码示例来了解如何管理SQLServer、AzureSQL和CosmosDB的连接器。6.1.1创建连接器SQLServer连接器创建SQLServer连接器时，需要指定服务器名称、数据库名称、用户名和密码。以下是一个创建SQLServer连接器的示例：{

"name":"SqlServerConnection",

"properties":{

"type":"SqlServer",

"typeProperties":{

"server":"<your_server_name>.",

"database":"<your_database_name>",

"authenticationType":"Basic",

"username":"<your_username>",

"password":"<your_password>"

},

"connectVia":{

"referenceName":"IntegrationRun