使用 Azure 数据工厂或 Azure Synapse Analytics 从/向 Salesforce 复制数据

项目
11/11/2024

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

提示

试用 Microsoft Fabric 中的数据工厂，这是一种适用于企业的一站式分析解决方案。 Microsoft Fabric 涵盖从数据移动到数据科学、实时分析、商业智能和报告的所有内容。了解如何免费开始新的试用！

本文概述如何使用 Azure 数据工厂和 Azure Synapse 管道中的复制活动从/向 Salesforce 复制数据。本文基于总体概述复制活动的复制活动概述一文。

重要

新的 Salesforce 连接器提供改进后的原生 Salesforce 支持。如果在解决方案中使用旧版 Salesforce 连接器，建议尽早升级 Salesforce 连接器。有关旧版和最新版本之间的差异的详细信息，请参阅此部分。

支持的功能

此 Salesforce 连接器支持以下功能：

支持的功能	IR
复制活动（源/接收器）	① ②
Lookup 活动	① ②

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

如需可以用作源或接收器的数据存储的列表，请参阅支持的数据存储表。

具体而言，Salesforce 连接器支持：

Salesforce 开发人员版、专业版、企业版或不受限制版。
从自定义域复制数据或将数据复制到自定义域（可以在生产环境和沙箱环境中配置自定义域）。

可以通过链接服务中的 apiVersion 属性显式设置用于读取/写入数据的 API 版本。将数据复制到 Salesforce 时，连接器使用 BULK API 2.0。

先决条件

在 Salesforce 中，必须启用 API 权限。
需要参考此官方文档或这篇文章建议中的分步指南在 Salesforce 门户中配置已连接的应用。
重要
- 执行用户必须具有“仅限 API”权限。
- 可以通过会话策略而非刷新令牌来更改访问令牌过期时间。

Salesforce Bulk API 2.0 限制

我们使用 Salesforce Bulk API 2.0 来查询和引入数据。在 Bulk API 2.0 中，会自动为你创建批处理。每个滚动 24 小时周期最多可提交 15,000 个批处理。如果批处理超出限制，则会失败。

在 Bulk API 2.0 中，只有引入作业使用批处理。查询作业不使用它。有关详细信息，请参阅如何在 Bulk API 2.0 中处理请求 - 开发人员指南。

有关详细信息，请参阅 Salesforce 开发人员限制中的“常规限制”部分。

开始使用

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

使用 UI 创建到 Salesforce 的链接服务

使用以下步骤在 Azure 门户 UI 中创建一个到 Salesforce 的链接服务。

浏览到 Azure 数据工厂或 Synapse 工作区中的“管理”选项卡并选择“链接服务”，然后单击“新建”：
- Azure 数据工厂
- Azure Synapse
搜索“Salesforce”并选择 Salesforce 连接器。
配置服务详细信息、测试连接并创建新的链接服务。

连接器配置详细信息

对于特定于 Salesforce 连接器的实体，以下部分提供有关用于定义这些实体的属性的详细信息。

链接服务属性

Salesforce 链接服务支持以下属性。

属性	描述	必需
type	类型属性必须设置为：SalesforceV2。	是
environmentUrl	指定 Salesforce 实例的 URL。例如，指定 `"https://<domainName>.my.salesforce.com"` 以从自定义域复制数据。参考此文章来了解如何配置或查看你的自定义域。	是
authenticationType	用于连接 Salesforce 的身份验证类型。允许的值为 OAuth2ClientCredentials。	是
clientId	指定 Salesforce OAuth 2.0 连接的应用的客户端 ID。有关详细信息，请参阅此文章	是
clientSecret	指定 Salesforce OAuth 2.0 连接的应用的客户端机密。有关详细信息，请参阅此文章	是
apiVersion	指定要使用的 Salesforce Bulk API 2.0 版本，例如 `52.0`。 Bulk API 2.0 仅支持 API 版本 >= 47.0。若要了解 Bulk API 2.0 版本，请参阅此文章。如果使用较低的 API 版本，则会出现失败。	是
connectVia	用于连接到数据存储的集成运行时。如果未指定，则使用默认 Azure Integration Runtime。	否

示例：存储凭据

{
    "name": "SalesforceLinkedService",
    "properties": {
        "type": "SalesforceV2",
        "typeProperties": {
            "environmentUrl": "<environment URL>",
            "authenticationType": "OAuth2ClientCredentials",
            "clientId": "<client ID>",
            "clientSecret": {
                "type": "SecureString",
                "value": "<client secret>"
            },
            "apiVersion": "<API Version>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

示例：在密钥保管库中存储凭据

{
    "name": "SalesforceLinkedService",
    "properties": {
        "type": "SalesforceV2",
        "typeProperties": {
            "environmentUrl": "<environment URL>",
            "authenticationType": "OAuth2ClientCredentials",
            "clientId": "<client ID>",
            "clientSecret": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of client secret in AKV>",
                "store":{
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                }
            },
            "apiVersion": "<API Version>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

示例：将凭据存储在密钥保管库中，以及 environmentUrl 和 clientId

通过在 Key Vault 以及 environmentUrl 和 clientId 中存储凭据，可以更长时间使用 UI 来编辑设置。必须选中“以 JSON 格式指定动态内容”复选框，并且必须手动进行此配置。此方案的优点是可以从 Key Vault 派生所有配置设置，而无需在此处进行任何的参数化。

{
    "name": "SalesforceLinkedService",
    "properties": {
        "type": "SalesforceV2",
        "typeProperties": {
            "environmentUrl": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of environment URL in AKV>",
                "store": {
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                },
            },
            "authenticationType": "OAuth2ClientCredentials",
            "clientId": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of client ID in AKV>",
                "store": {
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                },
            },
            "clientSecret": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of client secret in AKV>",
                "store":{
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                }
            },
            "apiVersion": "<API Version>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

数据集属性

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

要从/向 Salesforce 复制数据，请将数据集的 type 属性设置为 SalesforceV2Object。支持以下属性。

属性	描述	必需
type	类型属性必须设置为：SalesforceV2Object。	是
objectApiName	要从中检索数据的 Salesforce 对象名称。适用的自承载集成运行时版本为 5.44.8984.1 或更高版本。	对于源是“否”（如果指定源中的“query”），对于接收器为“是”
reportId	要从中检索数据的 Salesforce 报表的 ID。它在接收器中不受支持。使用报表时存在限制。适用的自承载集成运行时版本为 5.44.8984.1 或更高版本。	源（如果指定了源中的“query”），不支持接收器

重要

任何自定义对象均需要 API 名称的“__c”部分。

Salesforce 连接 API 名称

示例：

{
    "name": "SalesforceDataset",
    "properties": {
        "type": "SalesforceV2Object",
        "typeProperties": {
            "objectApiName": "MyTable__c"
        },
        "schema": [],
        "linkedServiceName": {
            "referenceName": "<Salesforce linked service name>",
            "type": "LinkedServiceReference"
        }
    }
}

复制活动属性

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

将 Salesforce 用作源类型

要从 Salesforce 复制数据，请将复制活动中的源类型设置为 SalesforceV2Source。复制活动的 source 节支持以下属性。

属性	描述	必需
type	复制活动源的 type 属性必须设置为 SalesforceV2Source。	是
查询	使用自定义查询读取数据。只能使用具有限制的 Salesforce 对象查询语言 (SOQL) 查询。有关 SOQL 限制，请参阅这篇文章。如果未指定查询，将检索在数据集的“objectApiName/reportId”中指定的 Salesforce 对象的所有数据。	否（如果指定了数据集中的“objectApiName/reportId”）
includeDeletedObjects	指示是查询现有记录，还是查询包括已删除记录在内的所有记录。如果未指定，则默认行为为 false。允许的值：false（默认）和 true。	否

重要

任何自定义对象均需要 API 名称的“__c”部分。

Salesforce 连接 API 名称列表

示例：

"activities":[
    {
        "name": "CopyFromSalesforce",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<Salesforce input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SalesforceV2Source",
                "query": "SELECT Col_Currency__c, Col_Date__c, Col_Email__c FROM AllDataType__c",
                "includeDeletedObjects": false
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

将 Salesforce 用作接收器类型

要向 Salesforce 复制数据，请将复制活动中的接收器类型设置为 SalesforceV2Sink。复制活动 sink 节支持以下属性。

属性	描述	必需
type	复制活动接收器的 type 属性必须设置为 SalesforceV2Sink。	是
writeBehavior	操作写入行为。允许的值为 Insert 和 Upsert。	否（默认值为 Insert）
externalIdFieldName	更新插入操作的外部的 ID 字段名称。指定的字段必须在 Salesforce 对象中定义为“外部 ID 字段”。它相应的输入数据中不能有 NULL 值。	对于“Upsert”是必需的
writeBatchSize	每批中写入到 Salesforce 的数据行计数。建议将此值设置为介于 10,000 和 200,000 之间的值。每批中的行太少会降低复制性能。每批中的行过多可能会导致 API 超时。	否（默认值为 100,000）
ignoreNullValues	指示是否忽略 NULL 值从输入数据期间写入操作。允许的值为 true 和 false。 - True：执行更新插入或更新操作时，保持目标对象中的数据不变。插入在执行插入操作时定义的默认值。 - False：执行更新插入或更新操作时，将目标对象中的数据更新为 NULL。执行插入操作时插入 NULL 值。	否（默认值为 false）
maxConcurrentConnections	活动运行期间与数据存储建立的并发连接的上限。仅在要限制并发连接时指定一个值。	无

示例：复制活动中的 Salesforce 接收器

"activities":[
    {
        "name": "CopyToSalesforce",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Salesforce output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SalesforceV2Sink",
                "writeBehavior": "Upsert",
                "externalIdFieldName": "CustomerId__c",
                "writeBatchSize": 10000,
                "ignoreNullValues": true
            }
        }
    }
]

Salesforce 的数据类型映射

从 Salesforce 复制数据时，使用以下映射从 Salesforce 数据类型内部映射到服务中的临时数据类型。若要了解复制活动如何将源架构和数据类型映射到接收器，请参阅架构和数据类型映射。

Salesforce 数据类型	服务临时数据类型
自动编号	String
复选框	布尔
货币	小数
Date	DateTime
日期/时间	DateTime
Email	String
ID	String
查找关系	String
多选择列表	String
Number	小数
百分比	小数
电话	String
选择列表	String
文本	String
文本区域	String
文本区域（长型值）	String
文本区域（丰富）	String
文本（加密）	String
URL	String

注意

Salesforce 的“数字”类型映射到 Azure 数据工厂中的“十进制”类型和 Azure Synapse 的管道即服务临时数据类型。 “十进制”类型遵循定义的精度和小数位数。 对于小数位数超过定义的小数位数的数据，其值会在预览数据和副本中进行舍入。 为了避免在 Azure 数据工厂和 Azure Synapse 管道中出现此类精度损失，请考虑在 Salesforce 的“自定义字段定义编辑”页中将小数位数提高到相当大的数值。

查找活动属性

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

升级 Salesforce 连接器

下面是帮助你升级 Salesforce 连接器的步骤：

参考先决条件在 Salesforce 门户中配置连接的应用。
新建 Salesforce 链接服务，并引用链接服务属性以对其进行配置。还需要手动更新依赖于旧链接服务的现有数据集，编辑每个数据集以改用新的链接服务。
如果你在复制活动源或引用旧版链接服务的查找活动中使用 SQL 查询，则需要将它们转换为 SOQL 查询。从 Salesforce 作为源类型和 Salesforce 对象查询语言 (SOQL) 详细了解 SOQL 查询。
readBehavior 在复制活动源或查找活动中被替换为 includeDeletedObjects。有关详细配置，请参阅将 Salesforce 用作源类型。

Salesforce 和 Salesforce 之间的差异（旧版）

Salesforce 连接器提供新功能，并与 Salesforce（旧版）连接器的大多数功能兼容。下表显示了 Salesforce 和 Salesforce（旧版）之间的功能差异。

Salesforce	Salesforce（旧版）
在 Salesforce Bulk API 2.0 中支持 SOQL。对于 SOQL 查询： • 不支持 GROUP BY、LIMIT、ORDER BY、OFFSET 或 TYPEOF 子句。 • 不支持 COUNT() 等聚合函数，可以使用 Salesforce 报表来实现它们。 • 不支持 GROUP BY 子句中的日期函数，但它们在 WHERE 子句中受支持。 • 不支持复合地址字段或复合地理位置字段。或者，查询复合字段的各个组成部分。 • 不支持父级到子级关系查询，而支持子级到父级关系查询。	支持 SQL 和 SOQL 语法。
指定查询时不支持包含二进制字段的对象。	指定查询时支持包含二进制字段的对象。
指定查询时支持批量 API 中的对象。	支持指定查询时支持批量 API 所不支持的对象。
通过选择报表 ID 来支持报表。	支持报表查询语法，如 `{call "<report name>"}`。

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

通过