Salesforce からデータを取り込む

このページでは、Salesforce からデータを取り込み、Databricks を使用してLakeFlow Connect に読み込む方法について説明します。

始める前に

インジェストパイプラインを作成するには、次の要件を満たす必要があります。

ワークスペースが Unity Catalog に対して有効になっています。
サーバレスコンピュートがワークスペースで有効になっています。サーバレスコンピュートの有効化を参照してください。
接続を作成する予定の場合: メタストアに対する CREATE CONNECTION 権限があります。
既存の接続を使用する予定の場合: 接続オブジェクトに対する USE CONNECTION 権限または ALL PRIVILEGES があります。
ターゲット・カタログに対する USE CATALOG 権限があります。
既存のスキーマに対する USE SCHEMA 権限と CREATE TABLE 権限、またはターゲットカタログに対する CREATE SCHEMA 権限を持っている。

Salesforce から取り込むには、次のことをお勧めします。

Databricks がデータを取得するために使用できる Salesforce ユーザーを作成します。ユーザーが API アクセスと、取り込む予定のすべてのオブジェクトにアクセスできることを確認します。

インジェストパイプラインを作成する

必要な権限: 接続 USE CONNECTION または ALL PRIVILEGES 。

この手順では、インジェストパイプラインを作成する方法について説明します。取り込まれた各テーブルは、同じ名前 (ただしすべて小文字) でストリーミングテーブルに書き込まれます。

Databricks UI
Databricks Asset Bundles
Databricks CLI

Databricksワークスペースのサイドバーで、 データ取り込み をクリックします。
データの追加 ページの Databricks コネクタ で、 Salesforce をクリックします。

Salesforce インジェストウィザードが開きます。
ウィザードの パイプライン ページで、インジェストパイプラインの一意の名前を入力します。
宛先カタログ ドロップダウンで、カタログを選択します。取り込まれたデータとイベントログは、このカタログに書き込まれます。
Salesforce データへのアクセスに必要な資格情報を保存する Unity Catalog 接続を選択します。

Salesforce 接続がない場合は、 接続の作成 をクリックします。メタストアに対する CREATE CONNECTION 特権が必要です。
パイプラインの作成および続行 をクリックします。
「 ソース 」ページで、取り込むテーブルを選択し、「次へ」をクリックします。

[ すべてのテーブル ] を選択した場合、Salesforce インジェストコネクタは、ソーススキーマ内のすべての既存および将来のテーブルを送信先スキーマに書き込みます。パイプラインごとに最大 250 個のオブジェクトがあります。
宛先ページで、書き込む Unity Catalog カタログとスキーマを選択します。

既存のスキーマを使用しない場合は、[ スキーマの作成 ] をクリックします。親カタログに対する USE CATALOG 権限と CREATE SCHEMA 権限が必要です。
パイプラインを保存と続行 をクリックします。
(オプション) 設定ページで、 スケジュールの作成 をクリックします。宛先テーブルを更新する頻度を設定します。
(オプション)パイプライン操作の成功または失敗のEメール通知を設定します。
パイプラインの保存と実行 をクリックします。

このタブでは、Databricks Asset Bundle を使用してインジェストパイプラインをデプロイする方法について説明します。バンドルには、ジョブとタスクの YAML 定義を含めることができ、 Databricks CLIを使用して管理され、異なるターゲットワークスペース (開発、ステージング、本番運用など) で共有および実行できます。詳細については、「アセットバンドルDatabricks」を参照してください。

Databricks CLI を使用して新しいバンドルを作成します。
Bash
```
databricks bundle init
```

バンドルに 2 つの新しいリソースファイルを追加します。

パイプライン定義ファイル (resources/sfdc_pipeline.yml)。
データ取り込みの頻度を制御するワークフローファイル (resources/sfdc_job.yml)。

次に、 resources/sfdc_pipeline.yml ファイルの例を示します。

YAML
variables:
  dest_catalog:
    default: main
  dest_schema:
    default: ingest_destination_schema

# The main pipeline for sfdc_dab
resources:
  pipelines:
    pipeline_sfdc:
      name: salesforce_pipeline
      catalog: ${var.dest_catalog}
      target: ${var.dest_schema}
      ingestion_definition:
        connection_name: <salesforce-connection>
        objects:
          # An array of objects to ingest from Salesforce. This example
          # ingests the AccountShare, AccountPartner, and ApexPage objects.
          - table:
              source_schema: objects
              source_table: AccountShare
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
          - table:
              source_schema: objects
              source_table: AccountPartner
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}
          - table:
              source_schema: objects
              source_table: ApexPage
              destination_catalog: ${var.dest_catalog}
              destination_schema: ${var.dest_schema}

次に、 resources/sfdc_job.yml ファイルの例を示します。

YAML
resources:
  jobs:
    sfdc_dab_job:
      name: sfdc_dab_job

      trigger:
        # Run this job every day, exactly one day from the last run
        # See https://docs.databricks.com/api/workspace/jobs/create#trigger
        periodic:
          interval: 1
          unit: DAYS

      email_notifications:
        on_failure:
          - <email-address>

      tasks:
        - task_key: refresh_pipeline
          pipeline_task:
            pipeline_id: ${resources.pipelines.pipeline_sfdc.id}

Databricks CLI を使用してパイプラインをデプロイします。
Bash
```
databricks bundle deploy
```

パイプライン定義で次のテーブル設定プロパティを使用して、取り込む特定の列を選択または選択解除できます。

include_columns: 必要に応じて、インジェストに含める列のリストを指定します。このオプションを使用して列を明示的に含めると、パイプラインは将来ソースに追加される列を自動的に除外します。将来の列を取り込むには、それらをリストに追加する必要があります。
exclude_columns: 必要に応じて、インジェストから除外する列のリストを指定します。このオプションを使用して列を明示的に除外すると、パイプラインには、今後ソースに追加される列が自動的に含まれます。将来の列を取り込むには、それらをリストに追加する必要があります。

パイプラインを作成するには:

databricks pipelines create --json "<pipeline-definition | json-file-path>"

パイプラインを更新するには:

databricks pipelines update --json "<pipeline-definition | json-file-path>"

パイプライン定義を取得するには:

databricks pipelines get "<pipeline-id>"

パイプラインを削除するには:

databricks pipelines delete "<pipeline-id>"

詳細については、以下を実行できます。

databricks pipelines --help
databricks pipelines <create|update|get|delete|...> --help

JSON パイプライン定義の例

JSON
"ingestion_definition": {

     "connection_name": "<connection-name>",

     "objects": [

       {

         "table": {

           "source_schema": "<source-schema>",

           "source_table": "<source-table>",

           "destination_catalog": "<destination-catalog>",

           "destination_schema": "<destination-schema>",

           "table_configuration": {

             "include_columns": ["<column-a>", "<column-b>", "<column-c>"]

           }

         }

       }

     ]

 }

パイプラインの開始、スケジュール設定、アラートの設定

パイプラインの詳細ページでパイプラインのスケジュールを作成できます。

パイプラインが作成されたら、 Databricks ワークスペースに再度アクセスし、[ パイプライン ] をクリックします。

新しいパイプラインがパイプラインリストに表示されます。
パイプラインの詳細を表示するには、パイプライン名をクリックします。
パイプラインの詳細ページで、 スケジュール をクリックしてパイプラインをスケジュールできます。
パイプラインに通知を設定するには、[ 設定 ] をクリックし、通知を追加します。

パイプラインに追加するスケジュールごとに、 LakeFlow Connect によってそのジョブが自動的に作成されます。インジェストパイプラインは、ジョブ内のタスクです。オプションで、ジョブにタスクを追加できます。

注記

パイプラインを実行すると、特定のテーブルに対して 2 つのソースビューが表示される場合があります。1 つのビューには、数式フィールドのスナップショットが含まれています。もう一方のビューには、数式以外のフィールドの増分データ取得が含まれています。これらのビューは、宛先テーブルで結合されます。

例: 2 つの Salesforce オブジェクトを別々のスキーマに取り込む

このセクションのパイプライン定義の例では、2 つの Salesforce オブジェクトを別々のスキーマに取り込みます。マルチデスティネーションパイプラインのサポートはAPIのみです。

YAML
resources:
  pipelines:
    pipeline_sfdc:
      name: salesforce_pipeline
      catalog: my_catalog_1 # Location of the pipeline event log
      schema: my_schema_1 # Location of the pipeline event log
      ingestion_definition:
        connection_name: <salesforce-connection>
        objects:
          - table:
              source_schema: objects
              source_table: AccountShare
              destination_catalog: my_catalog_1 # Location of this table
              destination_schema: my_schema_1 # Location of this table
          - table:
              source_schema: objects
              source_table: AccountPartner
              destination_catalog: my_catalog_2 # Location of this table
              destination_schema: my_schema_2 # Location of this table

例: 1 つの Salesforce オブジェクトを 3 回取り込む

このセクションのパイプライン定義の例では、Salesforce オブジェクトを 3 つの異なる宛先テーブルに取り込みます。マルチデスティネーションパイプラインのサポートはAPIのみです。

オプションで、取り込むテーブルの名前を変更できます。パイプライン内のテーブルの名前を変更すると、そのテーブルは API のみのパイプラインになり、UI でパイプラインを編集できなくなります。

YAML
resources:
  pipelines:
    pipeline_sfdc:
      name: salesforce_pipeline
      catalog: my_catalog_1	# Location of the pipeline event log
      schema: my_schema_1	# Location of the pipeline event log
      ingestion_definition:
        connection_name: <salesforce-connection>
        objects:
          - table:
              source_schema: objects
              source_table: Order
              destination_catalog: my_catalog_1	# Location of first copy
              destination_schema: my_schema_1	# Location of first copy
          - table:
              source_schema: objects
              source_table: Order
              destination_catalog: my_catalog_2	# Location of second copy
              destination_schema: my_schema_2	# Location of second copy
	        - table:
              source_schema: objects
              source_table: Order
              destination_catalog: my_catalog_2	# Location of third copy, renamed
              destination_schema: my_schema_2	# Location of third copy, renamed
              destination_table: order_duplicate # Table rename

始める前に​

インジェスト パイプラインを作成する​

JSON パイプライン定義の例​

パイプラインの開始、スケジュール設定、アラートの設定​

例: 2 つの Salesforce オブジェクトを別々のスキーマに取り込む​

例: 1 つの Salesforce オブジェクトを 3 回取り込む​

追加のリソース​