Salesforce からデータを取り込む

この記事では、LakeFlow Connect を使用して Salesforce からデータを取り込み、Databricks に読み込む方法について説明します。結果として得られるインジェスト パイプラインは Unity Catalog によって制御され、サーバレス コンピュートと DLT によって駆動されます。

Salesforce インジェスト コネクタは、次のソースをサポートしています。

• Salesforce Sales Cloud

始める前に

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

• ワークスペースが Unity Catalog に対して有効になっています。

• サーバレス コンピュートがワークスペースで有効になっています。 サーバレス コンピュートの有効化を参照してください。

• 接続を作成する予定の場合: メタストアに対する CREATE CONNECTION 権限があります。

  既存の接続を使用する予定の場合: 接続オブジェクトに対する USE CONNECTION 権限または ALL PRIVILEGES があります。

• ターゲット・カタログに対する USE CATALOG 権限があります。

• 既存のスキーマに対する USE SCHEMA 権限と CREATE TABLE 権限、またはターゲット カタログに対する CREATE SCHEMA 権限を持っている。

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

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

Salesforce 接続を作成する

必要なアクセス許可: メタストア CREATE CONNECTION 。これを許可するには、メタストア管理者に連絡してください。

既存の接続を使用してインジェストパイプラインを作成する場合は、次のセクションに進んでください。接続には USE CONNECTION または ALL PRIVILEGES が必要です。

Salesforce 接続を作成するには、次の手順を実行します。

1. Databricksワークスペースで、 カタログ > 外部ロケーション > 接続 > 接続の作成 をクリックします。

2. 接続名 で、Salesforce 接続の一意の名前を入力します。

3. 接続タイプ で Salesforce をクリックします。

4. Salesforce サンドボックス アカウントから取り込む場合は、 サンドボックスである を trueに設定します。

5. Salesforce でログイン をクリックします。

   

6. Salesforce サンドボックスから取り込む場合は、 カスタムドメインを使用 をクリックします。 サンドボックスのURLを入力し、ログインに進みます。 Databricks では、Databricks インジェスト専用の Salesforce ユーザーとしてログインすることをお勧めします。

   

   

7. 接続の作成 ページに戻ったら、 作成 をクリックします。

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

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

この手順では、インジェスト パイプラインを作成する方法について説明します。取り込まれた各テーブルは、明示的に名前を変更しない限り、デフォルトによって宛先に同じ名前 (ただしすべて小文字) のストリーミングテーブルに対応します。

Databricks UI

1. In the sidebar of the Databricks workspace, click Data Ingestion.

2. On the Add data page, under Databricks connectors, click Salesforce.

   The Salesforce ingestion wizard opens.

3. On the Pipeline page of the wizard, enter a unique name for the ingestion pipeline.

4. In the Destination catalog dropdown, select a catalog. Ingested data and event logs will be written to this catalog.

5. Select the Unity Catalog connection that stores the credentials required to access Salesforce data.

   If there are no Salesforce connections, click Create connection. You must have the CREATE CONNECTION privilege on the metastore.

6. Click Create pipeline and continue.

7. On the Source page, select the tables to ingest, and then click Next.

   If you select All tables, the Salesforce ingestion connector writes all existing and future tables in the source schema to Unity Catalog managed tables.

8. On the Destination page, select the Unity Catalog catalog and schema to write to.

   If you don’t want to use an existing schema, click Create schema. You must have the USE CATALOG and CREATE SCHEMA privileges on the parent catalog.

9. Click Save pipeline and continue.

10. On the Settings page, click Create schedule. Set the frequency to refresh the destination tables.

11. Optionally, set email notifications for pipeline operation success or failure.

12. Click Save and run pipeline.

Databricks Asset Bundles

This tab describes how to deploy an ingestion pipeline using Databricks Asset Bundles. Bundles can contain YAML definitions of jobs and tasks, are managed using the Databricks CLI, and can be shared and run in different target workspaces (such as development, staging, and production). For more information, see Databricks Asset Bundles.

You can use the following table configuration properties in your pipeline definition to select or deselect specific columns to ingest:

• include_columns: Optionally specify a list of columns to include for ingestion. If you use this option to explicitly include columns, the pipeline automatically excludes columns that are added to the source in the future. To ingest the future columns, you'll have to add them to the list.
• exclude_columns: Optionally specify a list of columns to exclude from ingestion. If you use this option to explicitly exclude columns, the pipeline automatically includes columns that are added to the source in the future. To ingest the future columns, you'll have to add them to the list.

1. Create a new bundle using the Databricks CLI:

   Bash

   databricks bundle init

2. Add two new resource files to the bundle:

   • A pipeline definition file (resources/sfdc_pipeline.yml).
   • A workflow file that controls the frequency of data ingestion (resources/sfdc_job.yml).

   The following is an example resources/sfdc_pipeline.yml file:

   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_configuration:
                   include_columns: # This can be exclude_columns instead
                     - <column_a>
                     - <column_b>
                     - <column_c>
             - 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}

   The following is an example resources/sfdc_job.yml file:

   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}

3. Deploy the pipeline using the Databricks CLI:

   Bash

   databricks bundle deploy

Databricks CLI

You can use the following table configuration properties in your pipeline definition to select or deselect specific columns to ingest:

• include_columns: Optionally specify a list of columns to include for ingestion. If you use this option to explicitly include columns, the pipeline automatically excludes columns that are added to the source in the future. To ingest the future columns, you'll have to add them to the list.
• exclude_columns: Optionally specify a list of columns to exclude from ingestion. If you use this option to explicitly exclude columns, the pipeline automatically includes columns that are added to the source in the future. To ingest the future columns, you'll have to add them to the list.

To create the pipeline:

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

To update the pipeline:

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

To get the pipeline definition:

databricks pipelines get "<pipeline-id>"

To delete the pipeline:

databricks pipelines delete "<pipeline-id>"

For more information, you can run:

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>"]

           }

         }

       }

     ]

 }

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

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

   新しいパイプラインがパイプライン リストに表示されます。

2. パイプラインの詳細を表示するには、パイプライン名をクリックします。

3. パイプラインの詳細ページで、 スケジュール をクリックしてパイプラインをスケジュールできます。

4. パイプラインに通知を設定するには、[ 設定 ] をクリックし、通知を追加します。

注記

パイプラインを実行すると、特定のテーブルに対して 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

追加のリソース

• コネクタの制限事項
• コネクタリファレンス
• コネクタに関する FAQ