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

适用于: Azure 数据工厂

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

支持的功能

以下活动支持此 Salesforce 连接器:

可以将数据从 Salesforce 复制到任何支持的接收器数据存储。 还可以将数据从任何支持的源数据存储复制到 Salesforce。 有关复制活动支持作为源或接收器的数据存储列表,请参阅支持的数据存储表。

具体而言,Salesforce 连接器支持:

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

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

备注

连接器不再设置 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 的链接服务。

  1. 浏览到 Azure 数据工厂中的“管理”选项卡并选择“链接服务”,然后单击“新建”:

    屏幕截图,显示如何使用 Azure 数据工厂 UI 创建新的链接服务。

  2. 搜索“Salesforce”并选择 Salesforce 连接器。

    Salesforce 连接器的屏幕截图。

  3. 配置服务详细信息、测试连接并创建新的链接服务。

    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"
        }
    }
}

数据集属性

有关可用于定义数据集的各部分和属性的完整列表,请参阅数据集一文。 本部分提供 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 复制数据,请将复制活动中的接收器类型设置为“SalesforceSink”。 复制活动 sink 节支持以下属性。

属性 说明 必需
type 复制活动接收器的 type 属性必须设置为 SalesforceSink
writeBehavior 操作写入行为。
允许的值为 InsertUpsert
否(默认值为 Insert)
externalIdFieldName 更新插入操作的外部的 ID 字段名称。 指定的字段必须在 Salesforce 对象中定义为“外部 ID 字段”。 它相应的输入数据中不能有 NULL 值。 对于“Upsert”是必需的
writeBatchSize 每批中写入到 Salesforce 的数据行计数。 否(默认值为5,000)
ignoreNullValues 指示是否忽略 NULL 值从输入数据期间写入操作。
允许的值为 truefalse
- 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"
日期时间格式 请参考此处的详细信息和下一部分中的示例。 请参考此处的详细信息和下一部分中的示例。
布尔值 表示为 FalseTrue,例如 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

查找活动属性

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

后续步骤

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