Salesforce資料來源 - DataWorks

Salesforce提供按需定製的軟體服務，這些服務涉及客戶關係管理的各個方面，例如從普通的連絡人管理、產品目錄到訂單管理、機會管理、銷售管理等。DataWorksData Integration支援讀取Salesforce類型的資料來源，本文為您介紹Salesforce的使用詳情。

支援的欄位類型

欄位類型	指令碼模式資料類型
address	STRING
anyType	STRING
base64	BYTES
boolean	BOOL
combobox	STRING
complexvalue	STRING
currency	DOUBLE
date	DATE
datetime	DATE
double	DOUBLE
email	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

進入建立頁面。
1. 前往Salesforce官網並登入。
2. 在頂部單擊，然後在左側導覽列單擊App Manager。
3. 在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。
1. 在App Manager列表中找到已建立的App，點擊App右側的，然後單擊View。
2. 在App詳情頁面，查看Consumer Key and Secret。
3. 複製Consumer Key和Consumer Secret值用於後續建立Saleforce資料來源時使用。

填寫資料來源參數

進入Data Integration頁面。
登入DataWorks控制台，切換至目標地區後，單擊左側導覽列的資料開發與治理 > Data Integration，在下拉框中選擇對應工作空間後單擊進入Data Integration。
單擊左側導覽列的資料來源，進入資料來源管理頁面。

單擊新增資料來源，搜尋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：查詢Salesforce Object。 query：通過SOQL查詢語句查詢資料。 bulk1：使用Salesforce Bulk API 1.0查詢Salesforce Object。 bulk2：使用Salesforce Bulk API 2.0查詢Salesforce Object。重要 bulk1、bulk2模式不支援address、geolocation等組合欄位類型。 bulk2模式不支援分布式任務。在某些情況下，bulk1的效能表現會比bulk2更好，您可以根據自己的Salesforce Object測試效能並選擇。	sobject
table	是	Salesforce Object，和表名類似，例如Account、Case、Group。當serviceType配置為sobject、bulk1或bulk2時必填。	無
beginDateTime	否	資料消費的開始、結束時間位點。當serviceType配置為sobject、bulk1或bulk2時使用。使用Salesforce Object的最後修改時間過濾資料，最後修改時間欄位會按照以下優先順序自動尋找：SystemModstamp>LastModifiedDate>CreatedDate。遵循左閉右開原則。 yyyymmddhhmmss格式的時間字串，您可以配合調度參數，實現增量抽取資料的效果。	無
endDateTime	否		無
splitPk	否	切分鍵。當serviceType配置為sobject時使用。 Salesforce Reader進行資料幫浦時，如果指定splitPk，表示您希望使用splitPk代表的欄位進行資料分區，資料同步因此會啟動並發任務進行資料同步，提高資料同步的效能。 splitPk支援datetime、int、long欄位，不符合這3個資料類型時，任務報錯。	無
blockCompoundColumn	否	組合資料類型行為。當serviceType配置為bulk1或bulk2時使用。取值： true：表示存在組合資料類型時，任務失敗，您需要去掉組合資料類型的列映射後重新運行。 false：表示將組合資料類型作為NULL值讀取。	true
bulkQueryJobTimeoutSeconds	否	批量資料準備逾時，單位：秒。當serviceType配置為bulk1或bulk2時使用。在開始讀取資料之前，Salesforce服務端需要先運行準備批量資料的任務，如果該任務已耗用時間超過本配置，則視為準備資料逾時，任務失敗。	86400
batchSize	否	單批下載條數。當serviceType配置為bulk1或bulk2時使用。本配置只需要比Salesforce批量資料準備任務自動分區的單批條數稍大即可獲得最高下載效能。資料下載是流式的，增大此配置項不會佔用更多記憶體。進階模式，嚮導模式不支援此參數的配置。	300000
where	否	篩選條件。當serviceType配置為sobject、bulk1或bulk2時使用。在實際業務情境中，可以進行資料篩選，例如Name != 'Aliyun'。如果不填寫where語句，資料同步均視作同步全量資料。不可以將where條件指定為limit 10，這不符合SOQL WHERE子句約束。	無
query	否	查詢語句。當serviceType配置為query時使用。在部分業務情境中，where配置項不足以描述所篩選的條件，您可以通過該配置項來自訂篩選SQL。配置該項後，資料同步系統會忽略table、column、beginDateTime、endDateTime、where和splitPk配置項，直接使用該項配置的內容對資料進行篩選。例如：`select Id, Name, IsDeleted from Account where Name!='Aliyun'`。進階模式，嚮導模式不支援此參數的配置。	無
queryAll	否	查詢語句，包含已刪除的資料。當serviceType配置為sobject或query時使用。配置為true時查詢包含已刪除的記錄，IsDeleted欄位可以用於區分記錄是否已刪除。	false
column	是	所配置的表中需要同步的列名集合，使用JSON的數組描述欄位資訊。支援列裁剪，可以挑選部分列進行匯出。支援列換序，可以不按照表schema資訊順序進行匯出。支援常量配置，如下： `[ { "name": "Id", "type": "STRING" }, { "name": "Name", "type": "STRING" }, { "name": "'123'", "type": "LONG" }, { "name": "'abc'", "type": "STRING" } ]` 說明 Id、Name為普通列名。 '123'為整型數字常量（注意需要加上一對單引號）。 'abc'為字串常量（注意需要加上一對單引號）。 column必須顯示指定同步的列集合，不允許為空白。	無
connectTimeoutSeconds	否	建立HTTP串連的逾時時間，單位：秒，超過本配置項時，任務失敗。進階模式，嚮導模式不支援此參數的配置。	30
socketTimeoutSeconds	否	HTTP串連失去響應，單位：秒，前後報文傳輸間隔大於此配置項時，任務失敗。進階模式，嚮導模式不支援此參數的配置。	600
retryIntervalSeconds	否	稍候再試，單位：秒。進階模式，嚮導模式不支援此參數的配置。	60
retryTimes	否	重試次數。進階模式，嚮導模式不支援此參數的配置。	3