すべてのプロダクト
Search
ドキュメントセンター

DataWorks:Salesforce データソース

最終更新日:Jan 11, 2025

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 にデータソースを追加する必要があります。詳細については、「データソースの追加と管理」をご参照ください。

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

  • Salesforce 公式: Salesforce の公式 Web サイトにログインして、Salesforce のアクセスアドレスを取得します。次に、アクセスアドレスに基づいてデータソースを追加します。

  • カスタム: 接続アプリケーションの [コンシューマーキー][コンシューマーシークレット] を指定して、Salesforce のアクセスアドレスを取得します。次に、アクセスアドレスに基づいてデータソースを追加します。

    カスタムモードでデータソースを追加する

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

    1. 作成ページに移動します。

      1. Salesforce 公式 Web サイト にログインします。

      2. 上部のナビゲーションバーで、image アイコンをクリックします。左側のナビゲーションペインで、[アプリ] > [アプリ マネージャー] を選択します。

      3. [lightning Experience アプリケーションマネージャー] ページで、[新規接続アプリケーション] をクリックします。

        image

    2. パラメーターを設定します。

      image

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

      領域

      説明

      1

      [接続アプリケーション名] フィールドに有効なアプリケーション名、[API 名] フィールドに有効な API 名、[連絡先 E メール] フィールドに E メールアドレスを、ビジネス要件に基づいて入力します。

      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、プロファイル、E メール、住所、電話番号)]

      • [一意のユーザー ID へのアクセス (openid)]

      • [API 経由のユーザーデータの管理 (api)]

      • [いつでもリクエストを実行 (更新トークン、offline_access)]

      4

      • [サポートされている認証フローのコード交換用証明キー (PKCE) 拡張機能を必須にする] をオフにします。

      • [web サーバーフローのシークレットを必須にする] を選択します。

      • [更新トークンフローのシークレットを必須にする] を選択します。

    3. 接続アプリケーションの [コンシューマーキーとコンシューマーシークレット] を表示します。

      1. [アプリケーションマネージャー] ページで作成したアプリケーションを見つけ、アプリケーションの右側にある image アイコンをクリックし、[表示] をクリックします。

      2. [接続アプリケーションの管理] ページの [API (OAuth 設定の有効化)] セクションで、[コンシューマーキーとコンシューマーシークレット] を表示します。

        image

      3. [コンシューマーキー][コンシューマーシークレット] をコピーします。

        image

    データソースの追加に必要なパラメーターの設定

    1. データ統合ページに移動します。

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

    2. 左側のナビゲーションペインで、[データソース] をクリックして、[データソース] ページに移動します。

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

      image

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

      パラメーター

      説明

      [ログインページ URL]

      フィールドに https://<Salesforce ドメイン名>/services/oauth2/authorize と入力します。

      [トークン署名 URL]

      フィールドに https://<Salesforce ドメイン名>/services/oauth2/token と入力します。

      [コンシューマーキー][コンシューマーシークレット]

      前の手順で取得した [コンシューマーキー][コンシューマーシークレット] を入力します。

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

      image

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

付録: コードとパラメーター

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

コードエディターを使用してバッチ同期タスクを設定する場合は、コードエディターの形式要件に基づいて、関連データソースのリーダーのパラメーターを設定する必要があります。形式要件の詳細については、「コードエディターを使用したバッチ同期タスクの設定」をご参照ください。以下の情報は、コードエディターのリーダーのパラメーターの設定の詳細について説明しています。

Salesforce Reader 用コード

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

{
  "type":"job",
  "version":"2.0",
  "steps":[
    {
      "stepType":"salesforce",
      "parameter":{
        "datasource":"", // データソース名
        "serviceType": "sobject", // 同期モード
        "table": "Account", // Salesforce オブジェクト
        "beginDateTime": "20230817184200", // 開始日時
        "endDateTime": "20231017184200", // 終了日時
        "where": "", // 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", // Bulk API 1.0 を使用
        "table": "Account", // Salesforce オブジェクト
        "beginDateTime": "20230817184200", // 開始日時
        "endDateTime": "20231017184200", // 終了日時
        "where": "", // 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", // Bulk API 2.0 を使用
        "table": "Account", // Salesforce オブジェクト
        "beginDateTime": "20230817184200", // 開始日時
        "endDateTime": "20231017184200", // 終了日時
        "where": "", // 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: Salesforce Object Query Language (SOQL) ステートメントを実行してデータをクエリする

{
  "type":"job",
  "version":"2.0",
  "steps":[
    {
      "stepType":"salesforce",
      "parameter":{
        "datasource":"", // データソース名
        "serviceType": "query", // SOQL ステートメントを使用
        "query": "select Id, Name, IsDeleted, CreatedDate from Account where Name!='Aliyun' ", // SOQL クエリ
        "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"
      }
    ]
  }
}

Salesforce Reader のコードのパラメーター

パラメーター

必須

説明

デフォルト値

datasource

はい

データソースの名前。追加されたデータソースの名前と同じである必要があります。コードエディターを使用してデータソースを追加できます。

デフォルト値なし

serviceType

いいえ

同期モード。有効な値:

  • sobject: Salesforce オブジェクトをクエリします。

  • query: SOQL ステートメントを実行してデータをクエリします。

  • bulk1: Salesforce Bulk API 1.0 を使用して Salesforce オブジェクトをクエリします。

  • bulk2: Salesforce Bulk API 2.0 を使用して Salesforce オブジェクトをクエリします。

    重要
    • bulk1 モードと bulk2 モードは、住所や地理位置情報データ型などの複合データ型の列をサポートしていません。

    • 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 に設定した場合、このパラメーターは必須です。

  • このパラメーターを設定すると、ソースオブジェクトはこのパラメーターの値に基づいてシャーディングされます。データ統合は並列スレッドを実行してデータを読み取ります。これにより、データの同期をより効率的に行うことができます。

  • このパラメーターは、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 に設定した場合、このパラメーターは必須です。

  • 実際のビジネスシナリオでは、WHERE 句を指定してデータをフィルタリングできます (例: Name != 'Aliyun')。

  • where パラメーターが指定されていないか空の場合、Salesforce Reader はすべてのデータを読み取ります。

  • where パラメーターを limit 10 に設定しないでください。これは、SOQL WHERE 句に対する Salesforce の制約に適合しません。

デフォルト値なし

query

いいえ

  • 絞り込んだデータフィルタリングに使用するクエリステートメント。 serviceType パラメーターを query に設定した場合、このパラメーターは必須です。

  • このパラメーターを設定すると、データ統合はこのパラメーターの値に基づいてデータをフィルタリングします。さらに、Salesforce Reader は、設定した 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リクエストのタイムアウト期間。単位: 秒。指定されたタイムアウト期間を超えると、タスクは失敗します。

  • このパラメーターは高度なパラメーターであり、コードエディターでのみ使用できます。

30

socketTimeoutSeconds

いいえ

  • HTTP レスポンスのタイムアウト期間。単位: 秒。2 つのパケットの間隔が指定されたタイムアウト期間より大きい場合、タスクは失敗します。

  • このパラメーターは高度なパラメーターであり、コードエディターでのみ使用できます。

600

retryIntervalSeconds

いいえ

  • 再試行間隔。単位: 秒。

  • このパラメーターは高度なパラメーターであり、コードエディターでのみ使用できます。

60

retryTimes

いいえ

  • 再試行回数。

  • このパラメーターは高度なパラメーターであり、コードエディターでのみ使用できます。

3