Salesforce提供按需定制的軟件服務,這些服務涉及客戶關系管理的各個方面,例如從普通的聯系人管理、產品目錄到訂單管理、機會管理、銷售管理等。DataWorks數據集成支持讀取Salesforce類型的數據源,本文為您介紹Salesforce的使用詳情。
支持的字段類型
字段類型 | 腳本模式數據類型 |
address | STRING |
anyType | STRING |
base64 | BYTES |
boolean | BOOL |
combobox | STRING |
complexvalue | STRING |
currency | DOUBLE |
date | DATE |
datetime | DATE |
double | DOUBLE |
STRING | |
encryptedstring | STRING |
id | STRING |
int | LONG |
json | STRING |
long | LONG |
multipicklist | STRING |
percent | DOUBLE |
phone | STRING |
picklist | STRING |
reference | STRING |
string | STRING |
textarea | STRING |
time | DATE |
url | STRING |
geolocation | STRING |
創建數據源
在進行數據同步任務開發時,您需要在DataWorks上創建一個對應的數據源,操作流程請參見創建并管理數據源,詳細的配置參數解釋可在配置界面查看對應參數的文案提示。
Salesforce支持如下方式創建數據源:
Salesforce官方:通過登錄Salesforce官網自動獲取Salesforce訪問地址信息,創建數據源。
自定義:通過指定您自己部署的Connected App的Consumer Key和Consumer Secret,自動獲取Salesforce訪問地址,創建數據源。
自定義模式數據源配置方式
創建Connected App
進入創建頁面。
前往Salesforce官網并登錄。
在頂部單擊
,然后在左側導航欄單擊App Manager。
在Lightning Experience App Manager頁面,單擊New Connected App。
填寫Connected App相關配置。
關鍵參數配置:
序號
參數說明
①
根據業務需求,創建符合Salesforce規范的應用名(Connected App Name)、API名稱(API Name)及聯系人郵箱(Contact Email)。
②
開啟Enable OAuth Settings。回調URL(Callback URL)填寫為
https://bff-cn-shanghai.data.aliyun.com/di/oauth/callback/index.html
。③
在Selected OAuth Scopes中添加以下Scope:
Access Connect REST API resources (chatter api)
Access the identity URL service (id, profile, email, address, phone)
Access unique user identifiers (openid)
Manage user data via APIs (api)
Perform requests at any time (refresh token, offline_access)
④
關閉Require Proof Key for Code Exchange (PKCE) Extension for Supported Authorization Flows。
開啟Require Secret for Web Server Flow。
開啟Require Secret for Refresh Token Flow。
查看Connected App的Consumer Key and Secret。
在App Manager列表中找到已創建的App,點擊App右側的
,然后單擊View。
在App詳情頁面,查看Consumer Key and Secret。
復制Consumer Key和Consumer Secret值用于后續創建Saleforce數據源時使用。
填寫數據源參數
進入數據集成頁面。
登錄DataWorks控制臺,切換至目標地域后,單擊左側導航欄的 ,在下拉框中選擇對應工作空間后單擊進入數據集成。
單擊左側導航欄的數據源,進入數據源管理頁面。
單擊新增數據源,搜索Salesforce并打開,在新增Salesforce數據源頁面,配置數據源類型為自定義。
關鍵參數配置:
參數
說明
登錄地址
填寫為
https://<Salesforce域名>/services/oauth2/authorize
。認證地址
填寫為
https://<Salesforce域名>/services/oauth2/token
。Consumer Key和Consumer Secret
填寫為已獲取Connected App詳情中的Consumer Key和Consumer Secret。
單擊登錄Salesforce,在授權頁面填寫用戶名、密碼后,單擊允許授權(Allow),完成數據源配置。
數據同步任務開發
數據同步任務的配置入口和通用配置流程可參見下文的配置指導。
單表離線同步任務配置指導
操作流程請參見通過向導模式配置離線同步任務、通過腳本模式配置離線同步任務。
腳本模式配置的全量參數和腳本Demo請參見下文的附錄:腳本Demo與參數說明。
附錄:腳本Demo與參數說明
離線任務腳本配置方式
如果您配置離線任務時使用腳本模式的方式進行配置,您需要按照統一的腳本格式要求,在任務腳本中編寫相應的參數,詳情請參見通過腳本模式配置離線同步任務,以下為您介紹腳本模式下數據源的參數配置詳情。
Reader腳本Demo
案例1:查詢Salesforce Object
{
"type":"job",
"version":"2.0",
"steps":[
{
"stepType":"salesforce",
"parameter":{
"datasource":"",
"serviceType": "sobject",
"table": "Account",
"beginDateTime": "20230817184200",
"endDateTime": "20231017184200",
"where": "",
"column": [
{
"type": "STRING",
"name": "Id"
},
{
"type": "STRING",
"name": "Name"
},
{
"type": "BOOL",
"name": "IsDeleted"
},
{
"type": "DATE",
"name": "CreatedDate"
}
]
},
"name":"Reader",
"category":"reader"
},
{
"stepType":"stream",
"parameter":{},
"name":"Writer",
"category":"writer"
}
],
"setting":{
"errorLimit":{
"record":"0"
},
"speed":{
"throttle":true,
"concurrent":1,
"mbps":"12"
}
},
"order":{
"hops":[
{
"from":"Reader",
"to":"Writer"
}
]
}
}
案例2:使用BULK API1.0查詢Salesforce Object
{
"type":"job",
"version":"2.0",
"steps":[
{
"stepType":"salesforce",
"parameter":{
"datasource":"",
"serviceType": "bulk1",
"table": "Account",
"beginDateTime": "20230817184200",
"endDateTime": "20231017184200",
"where": "",
"blockCompoundColumn":true,
"bulkQueryJobTimeoutSeconds":86400,
"column": [
{
"type": "STRING",
"name": "Id"
},
{
"type": "STRING",
"name": "Name"
},
{
"type": "BOOL",
"name": "IsDeleted"
},
{
"type": "DATE",
"name": "CreatedDate"
}
]
},
"name":"Reader",
"category":"reader"
},
{
"stepType":"stream",
"parameter":{
"print": true
},
"name":"Writer",
"category":"writer"
}
],
"setting":{
"errorLimit":{
"record":"0"
},
"speed":{
"concurrent":1
}
},
"order":{
"hops":[
{
"from":"Reader",
"to":"Writer"
}
]
}
}
案例3:使用BULK API2.0查詢Salesforce Object
{
"type":"job",
"version":"2.0",
"steps":[
{
"stepType":"salesforce",
"parameter":{
"datasource":"",
"serviceType": "bulk2",
"table": "Account",
"beginDateTime": "20230817184200",
"endDateTime": "20231017184200",
"where": "",
"blockCompoundColumn":true,
"bulkQueryJobTimeoutSeconds":86400,
"column": [
{
"type": "STRING",
"name": "Id"
},
{
"type": "STRING",
"name": "Name"
},
{
"type": "BOOL",
"name": "IsDeleted"
},
{
"type": "DATE",
"name": "CreatedDate"
}
]
},
"name":"Reader",
"category":"reader"
},
{
"stepType":"stream",
"parameter":{},
"name":"Writer",
"category":"writer"
}
],
"setting":{
"errorLimit":{
"record":"0"
},
"speed":{
"throttle":true,
"concurrent":1,
"mbps":"12"
}
},
"order":{
"hops":[
{
"from":"Reader",
"to":"Writer"
}
]
}
}
案例4:使用SOQL查詢語句
{
"type":"job",
"version":"2.0",
"steps":[
{
"stepType":"salesforce",
"parameter":{
"datasource":"",
"serviceType": "query",
"query": "select Id, Name, IsDeleted, CreatedDate from Account where Name!='Aliyun' ",
"column": [
{
"type": "STRING",
"name": "Id"
},
{
"type": "STRING",
"name": "Name"
},
{
"type": "BOOL",
"name": "IsDeleted"
},
{
"type": "DATE",
"name": "CreatedDate"
}
]
},
"name":"Reader",
"category":"reader"
},
{
"stepType":"stream",
"parameter":{},
"name":"Writer",
"category":"writer"
}
],
"setting":{
"errorLimit":{
"record":"0"
},
"speed":{
"throttle":true,
"concurrent":1,
"mbps":"12"
}
},
"order":{
"hops":[
{
"from":"Reader",
"to":"Writer"
}
]
}
}
Reader腳本參數
參數 | 是否必選 | 描述 | 默認值 |
datasource | 是 | 數據源名稱,腳本模式支持添加數據源,此配置項填寫的內容必須與添加的數據源名稱保持一致。 | 無 |
serviceType | 否 | 同步方式,支持以下配置項:
| sobject |
table | 是 | Salesforce Object,和表名類似,例如Account、Case、Group。當serviceType配置為sobject、bulk1或bulk2時必填。 | 無 |
beginDateTime | 否 |
| 無 |
endDateTime | 否 | 無 | |
splitPk | 否 |
| 無 |
blockCompoundColumn | 否 | 組合數據類型行為。當serviceType配置為bulk1或bulk2時使用。取值:
| true |
bulkQueryJobTimeoutSeconds | 否 |
| 86400 |
batchSize | 否 |
| 300000 |
where | 否 |
| 無 |
query | 否 |
| 無 |
queryAll | 否 |
| false |
column | 是 | 所配置的表中需要同步的列名集合,使用JSON的數組描述字段信息。
| 無 |
connectTimeoutSeconds | 否 |
| 30 |
socketTimeoutSeconds | 否 |
| 600 |
retryIntervalSeconds | 否 |
| 60 |
retryTimes | 否 |
| 3 |