PayPalからデータを取り込む

備考

ベータ版

この機能はベータ版です。使用するには、ワークスペース管理者は プレビュー ページから Lakeflow Connect for PayPal を有効にする必要があります。Databricksのプレビューを管理を参照してください。

このページでは、Lakeflow Connect を使用してマネージド PayPal インジェストパイプラインを作成する方法を示します。

要件

• 取り込み パイプラインを作成するには、まず次の要件を満たす必要があります。

  • ワークスペースでUnity Catalogが有効になっている必要があります。

  • Serverless コンピュートがワークスペースで有効になっている必要があります。Serverless コンピュートの要件を参照してください。

  • 新しい接続を作成するには、メタストアに対する CREATE CONNECTION 特権が必要です。Unity Catalog での特権の管理を参照してください。

    コネクタが UI ベースのパイプライン オーサリングをサポートしている場合、管理者はこのページのステップを完了することで、接続とパイプラインを同時に作成できます。 ただし、パイプラインを作成するユーザーが API ベースのパイプライン オーサリングを使用している場合、または管理者以外のユーザーである場合、管理者はまずカタログ エクスプローラーで接続を作成する必要があります。 「管理対象取り込みソースへの接続」を参照してください。

  • 既存の接続を使用するには、接続オブジェクトに対する USE CONNECTION 権限または ALL PRIVILEGES 権限が必要です。

  • ターゲットカタログに対する USE CATALOG 権限が必要です。

  • 既存のスキーマに対してUSE SCHEMAおよびCREATE TABLE権限、またはターゲットカタログに対してCREATE SCHEMA権限が必要です。

• PayPalから取り込むには、最初にDatabricksから認証を構成し、接続を作成します。「 PayPal への認証を構成する」および「 PayPal 接続を作成する」を参照してください。

コネクターのオプション

パイプラインスコープのオプションを source_configurations に設定します。使用例については、例をご覧ください。

オプション	スコープ	必須	適用対象	説明
start_datetime	パイプライン	No	すべてのインクリメンタルテーブル（transaction、dispute、invoice）	データの取り込みを開始する最も早い日時（yyyy-MM-ddTHH:mm:ssZ 形式）（例: 2024-01-01T00:00:00Z）。各テーブルの初回同期にのみ適用されます。以降の同期は保存されたカーソルから再開されます。defaultについては、コネクタのオプションをご覧ください。

オプション	スコープ	必須	適用対象	説明
start_datetime	パイプライン	No	すべてのインクリメンタルテーブル（transaction、dispute、invoice）	データの取り込みを開始する最も早い日時（yyyy-MM-ddTHH:mm:ssZ 形式）（例: 2024-01-01T00:00:00Z）。各テーブルの初回同期にのみ適用されます。以降の同期は保存されたカーソルから再開されます。defaultについては、コネクタのオプションをご覧ください。

取り込みパイプラインを作成する

サポートされているソーステーブルのリストについては、サポートされているソーステーブルを参照してください。

Databricks UI

1. Databricksワークスペースのサイドバーで、 データ取り込み をクリックします。
2. データの追加 ページの Databricks コネクタ で、 PayPal をクリックします。
3. インジェストウィザードの 接続 ページで、PayPal の認証情報を保存している接続を選択します。メタストアに対する CREATE CONNECTION 特権をお持ちの場合は、 接続を作成 をクリックして、PayPal の認証を構成の認証情報を使用する接続を作成します。
4. 次へ をクリックします。
5. インジェスト設定 ページで、パイプラインの名前を入力します。
6. イベント Logs を書き込むカタログとスキーマを選択します。カタログに対する USE CATALOG 権限と CREATE SCHEMA 権限を持っている場合、ドロップダウンメニューの スキーマの作成 をクリックしてスキーマを作成します。
7. パイプラインの作成および続行 をクリックします。
8. ソース ページで、取り込むテーブルを選択します。
9. 保存して続行 をクリックします。
10. 宛先 ページで、データを読み込むカタログとスキーマを選択します。カタログに対する USE CATALOG と CREATE SCHEMA の権限がある場合は、ドロップダウンメニューの スキーマの作成 をクリックしてスキーマを作成します。
11. 保存して続行 をクリックします。
12. (オプション) スケジュールと通知 ページで、 スケジュールの作成 をクリックします。宛先テーブルを更新する頻度を設定します。
13. (オプション) 通知の追加 をクリックして、パイプラインの操作が成功または失敗した場合にEメール通知を設定し、 保存してパイプラインをラン をクリックします。

Declarative Automation Bundles

宣言型オートメーションバンドルを使用して、PayPal パイプラインをコードとして管理します。バンドルには、ジョブとタスクの YAML 定義を含めることができ、Databricks CLI を使用して管理され、異なるターゲット ワークスペース（開発、ステージング、本番運用など）で共有およびランを実行できます。情報については、「宣言型オートメーションバンドルとは？」をご覧ください。

1. Databricks CLIを使用してバンドルを作成するには：

   Bash

   databricks bundle init

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

   • パイプライン定義ファイル (例えば、resources/paypal_pipeline.yml)。パイプライン.ingestion_definitionを参照してくださいおよび例。
   • データ取り込みの頻度を制御するジョブ定義ファイル（例：resources/paypal_job.yml）。

3. Databricks CLIを使用してパイプラインをデプロイします：

   Bash

   databricks bundle deploy

Databricks notebook

1. 次のノートブックをDatabricksワークスペースにインポートします：

ノートブックを新しいタブで開くOpen in Databricks

2. セル1とセル2はそのままにしておきます。変更しないでください。

3. セル3を、パイプライン構成の詳細に合わせて変更します。パイプライン.ingestion_definitionを参照してくださいおよび例。

4. 必要に応じて、パイプラインの詳細設定を行います。管理された取り込みパイプラインの一般的なパターンを参照してください。

5. 「 すべて実行 」をクリックします。

例

PayPalコネクタは、defaultソーススキーマで6つのソーステーブルを利用可能にします。個々のテーブルまたはスキーマ全体を取り込みます。

特定のテーブルのインジェスト

特定のテーブルのサブセットを取り込む場合、またはテーブルごとに宛先の名前をカスタマイズする場合は、このオプションを使用します。

Declarative Automation Bundles

次のパイプライン定義ファイルは、個々のPayPalテーブルを取り込みます：

YAML

resources:
  pipelines:
    paypal_pipeline:
      name: paypal_pipeline
      catalog: 'main'
      target: 'paypal_data'
      ingestion_definition:
        connection_name: paypal_connection
        objects:
          - table:
              source_schema: 'default'
              source_table: 'transaction'
              destination_catalog: 'main'
              destination_schema: 'paypal_data'
              destination_table: 'transaction'
          - table:
              source_schema: 'default'
              source_table: 'dispute'
              destination_catalog: 'main'
              destination_schema: 'paypal_data'
              destination_table: 'dispute'
          - table:
              source_schema: 'default'
              source_table: 'invoice'
              destination_catalog: 'main'
              destination_schema: 'paypal_data'
              destination_table: 'invoice'

Databricks notebook

次のパイプライン仕様は、個別の PayPal テーブルを取り込みます。

Python

pipeline_name = "paypal_pipeline"
connection_name = "<paypal-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "transaction",
          "destination_catalog": "main",
          "destination_schema": "paypal_data",
          "destination_table": "transaction"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "dispute",
          "destination_catalog": "main",
          "destination_schema": "paypal_data",
          "destination_table": "dispute"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "invoice",
          "destination_catalog": "main",
          "destination_schema": "paypal_data",
          "destination_table": "invoice"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

特定の起動日からの取り込み

インクリメンタルテーブルが最初の同期でどこまで遡って読み取るかを制御するには、start_datetime オプションを使用します。

Declarative Automation Bundles

以下のパイプライン定義ファイルは、指定された開始日からPayPalテーブルを取り込みます。

YAML

resources:
  pipelines:
    paypal_pipeline:
      name: paypal_pipeline
      catalog: 'main'
      target: 'paypal_data'
      ingestion_definition:
        connection_name: paypal_connection
        source_configurations:
          - api_source_connector_config:
              configs:
                start_datetime: '<start-datetime>'
        objects:
          - table:
              source_schema: 'default'
              source_table: 'transaction'
              destination_catalog: 'main'
              destination_schema: 'paypal_data'
              destination_table: 'transaction'
          - table:
              source_schema: 'default'
              source_table: 'dispute'
              destination_catalog: 'main'
              destination_schema: 'paypal_data'
              destination_table: 'dispute'
          - table:
              source_schema: 'default'
              source_table: 'invoice'
              destination_catalog: 'main'
              destination_schema: 'paypal_data'
              destination_table: 'invoice'

Databricks notebook

次のパイプライン仕様は、特定の起動日からPayPalテーブルをインジェストします：

Python

pipeline_name = "paypal_pipeline"
connection_name = "<paypal-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "source_configurations": [
      {
        "api_source_connector_config": {
          "configs": {
            "start_datetime": "<start-datetime>"
          }
        }
      }
    ],
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "transaction",
          "destination_catalog": "main",
          "destination_schema": "paypal_data",
          "destination_table": "transaction"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "dispute",
          "destination_catalog": "main",
          "destination_schema": "paypal_data",
          "destination_table": "dispute"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "invoice",
          "destination_catalog": "main",
          "destination_schema": "paypal_data",
          "destination_table": "invoice"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

スキーマ全体を取り込む

このオプションを使用して、すべてのPayPalソーステーブルを1つの宣言で単一の宛先スキーマに取り込みます。

Declarative Automation Bundles

次のパイプライン定義ファイルは、サポートされているすべての PayPal テーブルを宛先スキーマに取り込みます：

YAML

resources:
  pipelines:
    paypal_pipeline:
      name: paypal_pipeline
      catalog: 'main'
      target: 'paypal_data'
      ingestion_definition:
        connection_name: paypal_connection
        objects:
          - schema:
              source_schema: 'default'
              destination_catalog: 'main'
              destination_schema: 'paypal_data'

Databricks notebook

以下のパイプライン仕様は、サポートされているすべてのPayPalテーブルを送信先スキーマに取り込みます。

Python

pipeline_name = "paypal_pipeline"
connection_name = "<paypal-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "schema": {
          "source_schema": "default",
          "destination_catalog": "main",
          "destination_schema": "paypal_data"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

宣言型オートメーションバンドルのジョブ定義ファイル

以下は、Declarative Automation Bundlesで使用するジョブ定義ファイルの例です。ジョブは毎日実行されます。

Declarative Automation Bundles

YAML

resources:
  jobs:
    paypal_job:
      name: paypal_job
      schedule:
        quartz_cron_expression: '0 0 0 * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: paypal_ingestion
          pipeline_task:
            pipeline_id: ${resources.pipelines.paypal_pipeline.id}

一般的なパターン

高度なパイプライン構成については、マネージドインジェクションパイプラインの一般的なパターンを参照してください。

次のステップ

パイプラインの起動、スケジュール、アラートの設定。一般的なパイプラインのメンテナンス タスクを参照してください。

その他のリソース

• PayPalコネクタのFAQ
• PayPal コネクタの制限事項
• PayPal コネクタ リファレンス
• PayPalコネクターのトラブルシューティング