Salesforce データソース - DataWorks - Alibaba Cloud ドキュメントセンター

Salesforce は、連絡先管理、製品カタログ管理、注文管理、商談管理、および販売管理に重点を置いた顧客関係管理 (CRM) ソフトウェアを提供します。DataWorks は、Salesforce データソースからデータを読み取るための Salesforce Reader を提供します。このトピックでは、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 で同期タスクを開発する前に、「データソース管理」の指示に従って、必要なデータソースを DataWorks に追加する必要があります。データソースを追加する際に、DataWorks コンソールでパラメーターのツールチップを表示して、パラメーターの意味を理解できます。

Salesforce データソースは、次のいずれかの方法で作成できます。

[公式]: Salesforce 公式 Web サイトにログインして、Salesforce エンドポイントを自動的に取得し、データソースを作成します。

[カスタム]: DataWorks の専用接続認証情報を設定するには、[カスタム] モードを使用して [Salesforce] データソースを構成します。Salesforce システムにログインし、[接続アプリケーション] を作成して、セキュリティ認証情報 ([コンシューマーキー] と [コンシューマーシークレット]) を取得する必要があります。これらの認証情報は、DataWorks への接続権限を付与します。次の手順では、構成方法について説明します。

カスタムパターンデータソースの構成

接続アプリケーションの作成

作成ページに移動します。
1. Salesforce システムの URL に移動してログインします。
2. 上部のナビゲーションバーで、アイコンをクリックします。左側のナビゲーションウィンドウで、[アプリケーション] > [アプリケーションマネージャー] を選択します。
3. [Lightning Experience アプリケーションマネージャー] ページで、[新規接続アプリケーション] をクリックします。

[接続アプリケーション] を構成します。

次の表に、主要なパラメーターを示します。

序数	説明
1	[接続アプリケーション名] フィールドに有効なアプリケーション名を入力し、[API 参照名] フィールドに有効な API 名を入力し、必要に応じて [連絡先メール] フィールドにメールアドレスを入力します。
2	[OAuth 設定の有効化] を選択します。[コールバック URL] を `https://bff-cn-shanghai.data.aliyun.com/di/oauth/callback/index.html` に設定します。
3	[選択した OAuth 範囲] パラメーターに次の範囲を選択します。 Connect REST API リソースへのアクセス (chatter_api) ID URL サービスへのアクセス (id、profile、email、address、phone) 一意のユーザー識別子へのアクセス (openid) API を介したユーザーデータの管理 (api) いつでも要求を実行 (refresh_token、offline_access)
4	[サポートされている認証フローにコード交換用の証明キー (PKCE) 拡張を要求] をクリアします。 [Web サーバーフローのシークレットが必要] を選択します。 [更新トークンフローのシークレットが必要] を選択します。

接続アプリケーションの [コンシューマーキーとコンシューマーシークレット] を表示します。
1. [アプリケーションマネージャー] ページで作成したアプリケーションを見つけ、アプリケーションの右側にあるアイコンをクリックし、[表示] をクリックします。
2. [接続アプリケーションを管理] ページの [API (OAuth 設定の有効化)] セクションで、[コンシューマーキーとコンシューマーシークレット] を表示します。
3. [コンシューマーキー] と [コンシューマーシークレット] をコピーします。

データソースの追加に必要なパラメーターの構成

データ統合ページに移動します。
DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、[データ統合] > [データ統合] を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ統合に移動] をクリックします。
左側のナビゲーションウィンドウで、[データソース] をクリックして [データソース] ページに移動します。

[データソース] ページで、[データソースの追加] をクリックします。[データソースの追加] ダイアログボックスで、Salesforce を検索して Salesforce をクリックします。[Salesforce データソースの追加] ダイアログボックスで、[データソースタイプ] パラメーターに [カスタム] を選択します。

次の表に、主要なパラメーターを示します。

パラメーター	説明
ログインページ URL	`https://<Salesforce ドメイン名>/services/oauth2/authorize` と入力します。
認証アドレス	`https://<Salesforce ドメイン名>/services/oauth2/token` と入力します。
コンシューマーキーとコンシューマーシークレット	前の手順で取得した [コンシューマーキー] と [コンシューマーシークレット] を入力します。

[Salesforce にログイン] をクリックします。表示されたページで、ユーザー名とパスワードを入力し、[許可] をクリックします。

重要

Salesforce はサードパーティのサービスです。リソースグループにアタッチされている VPC ネットワークが Salesforce プラットフォームに接続できることを確認してください。そうでない場合、データソースは作成できません。

データ同期タスクの開発

同期タスクの構成のエントリポイントと手順については、次の構成ガイドをご参照ください。

単一テーブルのデータを同期するためのバッチ同期タスクの構成

構成手順の詳細については、「コードレス UI での構成」および「コードエディタでの構成」をご参照ください。
コードエディタのすべてのパラメーターとコードサンプルについては、「付録: スクリプトのデモとパラメーターの説明」をご参照ください。

付録: スクリプトのデモとパラメーターの説明

コードエディタを使用したバッチ同期タスクの構成

コードエディタを使用してバッチ同期タスクを構成する場合、統一されたスクリプト形式の要件に基づいて、スクリプト内の関連パラメーターを構成する必要があります。詳細については、「コードエディタでタスクを構成する」をご参照ください。次の情報では、コードエディタを使用してバッチ同期タスクを構成する際に、データソースに対して構成する必要があるパラメーターについて説明します。

Reader スクリプトのデモ

例 1: Salesforce オブジェクトのクエリ

{
  "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 API 1.0 を使用した Salesforce オブジェクトのクエリ

{
  "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 API 2.0 を使用した Salesforce オブジェクトのクエリ

{
  "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 オブジェクトをクエリします。 query: SOQL 文を実行してデータをクエリします。 bulk1: Salesforce Bulk API 1.0 を使用して Salesforce オブジェクトをクエリします。 bulk2: Salesforce Bulk API 2.0 を使用して Salesforce オブジェクトをクエリします。重要 bulk1 および bulk2 モードは、address や geolocation などの複合フィールド型をサポートしていません。 bulk2 モードは分散タスクをサポートしていません。場合によっては、bulk1 の方が bulk2 よりもクエリパフォーマンスが優れています。Salesforce オブジェクトに基づいてクエリパフォーマンスをテストし、必要に応じて同期モードを選択できます。	sobject
table	はい	Salesforce オブジェクト (Account、Case、Group など)。オブジェクトはテーブルと同等です。serviceType パラメーターを sobject、bulk1、または bulk2 に設定した場合、このパラメーターは必須です。	なし
beginDateTime	いいえ	データ消費の開始時刻と終了時刻。serviceType パラメーターを sobject、bulk1、または bulk2 に設定した場合、これらのパラメーターは必須です。 Salesforce オブジェクトの最終変更時刻に基づいてデータをフィルタリングする場合、オブジェクトの変更に関連する時間フィールドは、SystemModstamp > LastModifiedDate > CreatedDate の順序でクエリされます。時間範囲は、左側が閉じて右側が開いている区間です。時刻は yyyymmddhhmmss 形式です。これらのパラメーターを DataWorks のスケジューリングパラメーターと組み合わせて使用して、増分データを読み取ることができます。	なし
endDateTime	いいえ		なし
splitPk	いいえ	データシャーディングに使用されるフィールド。serviceType パラメーターを sobject に設定した場合、このパラメーターは必須です。 Salesforce Reader がデータを抽出する際に splitPk を指定すると、指定されたフィールドがデータシャーディングに使用されます。その後、Data Integration は同時タスクを開始してデータを同期し、同期効率を向上させます。このパラメーターは、datetime、int、または long データ型のフィールドに設定できます。他の型のフィールドでは、エラーが報告されます。	なし
blockCompoundColumn	いいえ	複合データ型の列をサポートするかどうかを指定します。serviceType パラメーターを bulk1 または bulk2 に設定した場合、このパラメーターは必須です。有効な値: true: 複合データ型の列が存在する場合、タスクは失敗し、複合データ型の列マッピングを削除した後に再実行する必要があります。 false: 複合データ型の列のデータは NULL 値として読み取られます。	true
bulkQueryJobTimeoutSeconds	いいえ	バッチデータ準備のタイムアウト期間。単位: 秒。serviceType パラメーターを bulk1 または bulk2 に設定した場合、このパラメーターは必須です。 Salesforce Reader がデータの読み取りを開始する前に、Salesforce サーバーはバッチデータを準備するタスクを実行します。タスクの実行時間が指定されたタイムアウト期間を超えると、タスクはタイムアウトして失敗します。	86400
batchSize	いいえ	一度にダウンロードするデータレコードの数。serviceType パラメーターを bulk1 または bulk2 に設定した場合、このパラメーターは必須です。最適なダウンロードパフォーマンスを実現するには、このパラメーターを Salesforce のバッチデータ準備タスクの自動シャーディングの値よりわずかに大きい値に設定できます。データはストリーミングモードでダウンロードされます。したがって、このパラメーターの値を増やしても、より多くのメモリを占有することはありません。このパラメーターは詳細パラメーターであり、コードエディタでのみ使用できます。	300000
where	いいえ	WHERE 句。serviceType パラメーターを sobject、bulk1、または bulk2 に設定した場合、このパラメーターは必須です。実際のビジネスシナリオでは、Name != 'Aliyun' などの WHERE 句を指定してデータをフィルタリングできます。 WHERE 句を省略すると、完全なデータ同期が行われます。 where パラメーターを limit 10 に設定しないでください。これは、SOQL WHERE 句に対する Salesforce の制約に準拠していません。	なし
query	いいえ	詳細なデータフィルタリングに使用されるクエリ文。serviceType パラメーターを query に設定した場合、このパラメーターは必須です。一部のビジネスシナリオでは、where パラメーターではフィルター条件を記述するのに不十分です。このパラメーターを使用して、フィルター SQL 文をカスタマイズできます。このパラメーターを構成すると、データ同期システムは table、column、beginDateTime、endDateTime、where、および splitPk パラメーターを無視し、このパラメーターの内容を直接使用してデータをフィルタリングします。例: `select Id, Name, IsDeleted from Account where Name!='Aliyun'`。このパラメーターは詳細パラメーターであり、コードエディタでのみ使用できます。	なし
queryAll	いいえ	すべてのデータをクエリするために使用される SQL 文。serviceType パラメーターを sobject または query に設定した場合、このパラメーターは必須です。このパラメーターを true に設定すると、削除されたデータを含むすべてのデータがクエリされます。IsDeleted フィールドを使用して、データレコードが削除されたかどうかを判断できます。	false
column	はい	同期する列の名前。JSON 配列で名前を指定します。列のサブセットを選択してエクスポートできます。列の順序は変更できます。これは、ソーステーブルのスキーマで指定された順序とは異なる順序で列を指定できることを示します。定数がサポートされています。例: `[ { "name": "Id", "type": "STRING" }, { "name": "Name", "type": "STRING" }, { "name": "'123'", "type": "LONG" }, { "name": "'abc'", "type": "STRING" } ]` 説明 Id と Name: 列名。 '123': 整数定数。単一引用符 (') で囲まれています。 'abc': 文字列定数。単一引用符 (') で囲まれています。 column パラメーターは、データを読み取るすべての列を明示的に指定する必要があります。このパラメーターを空にすることはできません。	なし
connectTimeoutSeconds	いいえ	HTTP リクエストのタイムアウト期間。単位: 秒。指定されたタイムアウト期間を超えると、タスクは失敗します。このパラメーターは、コードレス UI ではなく、詳細モードでのみ構成できます。	30
socketTimeoutSeconds	いいえ	HTTP 応答のタイムアウト期間。単位: 秒。2 つのパケット間の間隔が指定されたタイムアウト期間より大きい場合、タスクは失敗します。このパラメーターは詳細パラメーターであり、コードエディタでのみ使用できます。	600
retryIntervalSeconds	いいえ	再試行間隔。単位: 秒。このパラメーターは詳細モード専用であり、コードレス UI ではサポートされていません。	60
retryTimes	いいえ	再試行回数。この詳細モードパラメーターは、コードレス UI では構成できません。	3