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

项目
04/12/2024

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

提示

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

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

重要

服务发布了一款新的 Salesforce 连接器，该连接器提供了更好的原生 Salesforce 支持，详情请参阅 Salesforce 连接器一文。

支持的功能

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

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

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

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

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

Salesforce 开发人员版、专业版、企业版或不受限制版。
从/向 Salesforce 生产、沙盒和自定义域复制数据。

注意

此功能支持从上述 Salesforce 环境中的任何架构副本，包括非营利成功包 (NPSP)。

Salesforce 连接器在 Salesforce REST/Bulk API 之上构建。从 Salesforce 复制数据时，连接器会根据数据大小在 REST 和批量 API 之间自动选择。当结果集较大时，使用批量 API 来提高性能。你可通过链接服务中的 apiVersion 属性显式设置用于读取/写入数据的 API 版本。将数据复制到 Salesforce 时，连接器使用 BULK API v1。

注意

连接器不再设置 Salesforce API 的默认版本。为了向后兼容，如果之前设置了默认 API 版本，它将一直工作。对于源，默认值为 45.0，对于接收器，默认值为 40.0。

先决条件

在 Salesforce 中，必须启用 API 权限。

Salesforce 请求限制

Salesforce 对 API 请求总数和并发 API 请求均有限制。请注意以下几点：

如果并发请求数超过限制，则将进行限制并且会看到随机失败。
如果请求总数超过限制，会阻止 Salesforce 帐户 24 小时。

在这两种情况下，还可能会收到“REQUEST_LIMIT_EXCEEDED”错误消息。有关详细信息，请参阅 Salesforce 开发人员限制中的“API 请求限制”部分。

入门

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

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

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

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

连接器配置详细信息

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

链接服务属性

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

属性	描述	必需
type	类型属性必须设置为 Salesforce。	是
environmentUrl	指定 Salesforce 实例的 URL。 - 默认为 `"https://login.salesforce.com"`。 - 要从沙盒复制数据，请指定 `"https://test.salesforce.com"`。 - 要从自定义域复制数据，请指定 `"https://[domain].my.salesforce.com"`（以此为例）。	否
username	为用户帐户指定用户名。	是
password	指定用户帐户的密码。将此字段标记为 SecureString 以安全地存储它，或引用 Azure Key Vault 中存储的机密。	是
securityToken	为用户帐户指定安全令牌。若要了解有关安全令牌的一般信息，请参阅 Security and the API（安全性和 API）。仅当将 Integration Runtime 的 IP 添加到 Salesforce 上的受信任 IP 地址列表时，才能跳过安全令牌。使用 Azure IR 时，请参阅 Azure Integration Runtime IP 地址。有关如何获取和重置安全令牌的说明，请参阅获取安全令牌。将此字段标记为 SecureString 以安全地存储它，或引用 Azure Key Vault 中存储的机密。	否
apiVersion	指定要使用的 Salesforce REST/Bulk API 版本，例如 `52.0`。	否
connectVia	用于连接到数据存储的集成运行时。如果未指定，则使用默认 Azure Integration Runtime。	否

示例：存储凭据

{
    "name": "SalesforceLinkedService",
    "properties": {
        "type": "Salesforce",
        "typeProperties": {
            "username": "<username>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            },
            "securityToken": {
                "type": "SecureString",
                "value": "<security token>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

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

{
    "name": "SalesforceLinkedService",
    "properties": {
        "type": "Salesforce",
        "typeProperties": {
            "username": "<username>",
            "password": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of password in AKV>",
                "store":{
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                }
            },
            "securityToken": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of security token in AKV>",
                "store":{
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                }
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

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

请注意，这样一来，你将不能再使用 UI 编辑设置。将选中“以 JSON 格式指定动态内容”复选框，并且你必须完全手动编辑此配置。优点是可以从密钥保管库派生所有配置设置，而无需在此处进行任何的参数化。

{
    "name": "SalesforceLinkedService",
    "properties": {
        "type": "Salesforce",
        "typeProperties": {
            "environmentUrl": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of environment URL in AKV>",
                "store": {
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                },
            },
            "username": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of username in AKV>",
                "store": {
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                },
            },
            "password": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of password in AKV>",
                "store":{
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                }
            },
            "securityToken": {
                "type": "AzureKeyVaultSecret",
                "secretName": "<secret name of security token in AKV>",
                "store":{
                    "referenceName": "<Azure Key Vault linked service>",
                    "type": "LinkedServiceReference"
                }
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

数据集属性

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

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

属性	描述	必需
type	type 属性必须设置为 SalesforceObject。	是
objectApiName	要从中检索数据的 Salesforce 对象名称。	对于源为“No”，对于接收器为“Yes”

重要

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

显示 Salesforce 连接 API 名称的屏幕截图。

示例：

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

注意

为了向后兼容：从 Salesforce 复制数据时，如果使用以前的“RelationalTable”类型数据集，它会在你看到切换到新的“SalesforceObject”类型的建议时继续工作。

属性	描述	必需
type	数据集的 type 属性必须设置为 RelationalTable。	是
tableName	在 Salesforce 中表的名称。	否（如果指定了活动源中的“query”）

复制活动属性

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

将 Salesforce 用作源类型

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

属性	描述	必需
type	复制活动源的 type 属性必须设置为 SalesforceSource。	是
查询	使用自定义查询读取数据。可以使用 Salesforce 对象查询语言 (SOQL) 查询或 SQL-92 查询。请在查询提示部分中查看更多提示。如果未指定查询，将检索在数据集的“objectApiName”中指定的 Salesforce 对象的所有数据。	否（如果指定了数据集中的“objectApiName”）
readBehavior	指示是查询现有记录，还是查询包括已删除记录在内的所有记录。如果未指定，默认行为是前者。允许的值：query（默认值）、queryAll。	否

重要

任何自定义对象均需要 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": "SalesforceSource",
                "query": "SELECT Col_Currency__c, Col_Date__c, Col_Email__c FROM AllDataType__c"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

注意

为了向后兼容：从 Salesforce 复制数据时，如果使用以前的“RelationalSource”类型副本，则当你看到切换到新的“SalesforceSource”类型的建议时，该源会继续工作。

注意

Salesforce 源不支持自承载集成运行时中的代理设置，但接收器支持。

将 Salesforce 用作接收器类型

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

属性	描述	必需
type	复制活动接收器的 type 属性必须设置为 SalesforceSink。	是
writeBehavior	操作写入行为。允许的值为 Insert 和 Upsert。	否（默认值为 Insert）
externalIdFieldName	更新插入操作的外部的 ID 字段名称。指定的字段必须在 Salesforce 对象中定义为“外部 ID 字段”。它相应的输入数据中不能有 NULL 值。	对于“Upsert”是必需的
writeBatchSize	每批中写入到 Salesforce 的数据行计数。	否（默认值为5,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": "SalesforceSink",
                "writeBehavior": "Upsert",
                "externalIdFieldName": "CustomerId__c",
                "writeBatchSize": 10000,
                "ignoreNullValues": true
            }
        }
    }
]

查询提示

从 Salesforce 报表检索数据

可通过将查询指定为 {call "<report name>"} 从 Salesforce 报表检索数据。例如 "query": "{call \"TestReport\"}"。

从 Salesforce 回收站中检索删除的记录

若要从 Salesforce 回收站中查询软删除的记录，需要将 readBehavior 指定为 queryAll。

SOQL 与 SQL 查询语法之间的差异

从 Salesforce 中复制数据时，可以使用 SOQL 查询或 SQL 查询。请注意，这两者具有不同的语法和功能支持，不要混用。建议使用 Salesforce 原本就支持的 SOQL 查询。下表列出了主要差异：

语法	SOQL 模式	SQL 模式
列选择	需要枚举要在查询中复制的字段，例如 `SELECT field1, filed2 FROM objectname`	除了列选择之外，还支持 `SELECT *`。
引号	字段/对象名称不能用引号引起来。	字段/对象名称可以用引号引起来，例如 `SELECT "id" FROM "Account"`
日期时间格式	请参考此处的详细信息和下一部分中的示例。	请参考此处的详细信息和下一部分中的示例。
布尔值	表示为 `False` 和 `True`，例如 `SELECT … WHERE IsDeleted=True`。	表示为 0 或 1，例如 `SELECT … WHERE IsDeleted=1`。
列重命名	不支持。	支持，例如：`SELECT a AS b FROM …`。
关系	支持，例如 `Account_vod__r.nvs_Country__c`。	不支持。

使用 DateTime 列上的 where 子句检索数据

当指定 SOQL 或 SQL 查询时，请注意 DateTime 的格式差异。例如：

SOQL 示例：SELECT Id, Name, BillingCity FROM Account WHERE LastModifiedDate >= @{formatDateTime(pipeline().parameters.StartTime,'yyyy-MM-ddTHH:mm:ssZ')} AND LastModifiedDate < @{formatDateTime(pipeline().parameters.EndTime,'yyyy-MM-ddTHH:mm:ssZ')}
SQL 示例：SELECT * FROM Account WHERE LastModifiedDate >= {ts'@{formatDateTime(pipeline().parameters.StartTime,'yyyy-MM-dd HH:mm:ss')}'} AND LastModifiedDate < {ts'@{formatDateTime(pipeline().parameters.EndTime,'yyyy-MM-dd HH:mm:ss')}'}

MALFORMED_QUERY:Truncated 错误

如果遇到“MALFORMED_QUERY:Truncated”错误，通常是因为在数据中存在 JunctionIdList 类型列，而 Salesforce 在支持此类具有大量行的数据方面存在限制。若要缓解这种情况，请尝试排除 JunctionIdList 列或限制要复制的行数（可以将其划分为多个复制活动运行）。

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 活动。

后续步骤

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