Anysphere Organizationからデータを取り込む

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

このページでは、Lakeflow Connect を使用して管理された Anysphere Organization インジェスト パイプラインを作成する方法について説明します。

要件

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

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

  • ワークスペースで Serverless コンピュートを有効にする必要があります。See Serverless コンピュート requirements.

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

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

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

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

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

• Anysphere Organization から取り込むには、まず Databricks から認証を構成する必要があります。Anysphere Organization への認証の構成を参照してください。

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

Databricks UI

1. Databricksワークスペースのサイドバーで、 データ取り込み をクリックします。
2. On the Add data page, under Databricks connectors , click Anysphere Organization .
3. On the Connection page of the ingestion wizard, select the connection that stores your Anysphere Organization access credentials.If you have the CREATE CONNECTION privilege on the metastore, you can click 接続の作成 to create a new connection with the authentication details in Create an Anysphere Organization connection.
4. 次へ をクリックします。
5. インジェスト セットアップ ページで、パイプラインの一意の名前を入力します。
6. イベント event Logs を書き込むカタログとスキーマを選択します。カタログに対する USE CATALOG 権限と CREATE SCHEMA 権限がある場合は、ドロップダウン メニューの [スキーマの作成] をクリックして新しいスキーマを作成できます。
7. パイプラインの作成および続行 をクリックします。
8. ソース ページで、取り込むテーブルを選択します。
9. 保存して続行 をクリックします。
10. 宛先 ページで、データをロードするカタログとスキーマを選択します。カタログに対する USE CATALOG 権限と CREATE SCHEMA 権限がある場合は、ドロップダウンメニューで スキーマの作成 をクリックして新しいスキーマを作成できます。
11. 保存して続行 をクリックします。
12. (オプション) スケジュールと通知 ページで、 スケジュールの作成 をクリックします。移行先テーブルを更新する頻度を設定します。
13. (オプション) 通知の追加 をクリックしてパイプライン操作の成功または失敗に対するEメール通知を設定し、 [パイプラインの保存とラン] をクリックします。

Declarative Automation Bundles

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

1. Databricks CLI を使用してバンドルを作成するには、次の手順を実行します。

   Bash

   databricks bundle init

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

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

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

   Bash

   databricks bundle deploy

Databricks notebook

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

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

2. セル1とセル2はそのままで、変更しないでください。変更しないでください。

3. パイプライン構成の詳細を使用してセル 3 を変更します。パイプライン.ingestion_definition を参照してください。また、 例を参照してください。

4. 高度なパイプライン設定を必要に応じて構成します。マネージド取り込みパイプラインの一般的なパターンについては、Common patterns for managed ingestion pipelines を参照してください。

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

例

これらの例を使用して、パイプラインを構成します。

スキーマ全体を取り込む

Declarative Automation Bundles

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

YAML

resources:
  pipelines:
    anysphere_organization_pipeline:
      name: anysphere_organization_pipeline
      catalog: 'main'
      target: 'anysphere_organization_data'
      ingestion_definition:
        connection_name: anysphere_organization_connection
        objects:
          - schema:
              source_schema: 'default'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'

Databricks notebook

次のパイプライン仕様では、サポートされているすべての Anysphere Organization テーブルが宛先スキーマに取り込まれます。

Python

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

特定のテーブルを取り込む

個別の Anysphere Organization テーブルを取り込みます。サポートされているテーブルの一覧については、サポートされているソーステーブルを参照してください。

Declarative Automation Bundles

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

YAML

resources:
  pipelines:
    anysphere_organization_pipeline:
      name: anysphere_organization_pipeline
      catalog: 'main'
      target: 'anysphere_organization_data'
      ingestion_definition:
        connection_name: anysphere_organization_connection
        objects:
          - table:
              source_schema: 'default'
              source_table: 'organization_groups'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'
              destination_table: 'organization_groups'
          - table:
              source_schema: 'default'
              source_table: 'organization_group_members'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'
              destination_table: 'organization_group_members'
          - table:
              source_schema: 'default'
              source_table: 'organization_members'
              destination_catalog: 'main'
              destination_schema: 'anysphere_organization_data'
              destination_table: 'organization_members'

Databricks notebook

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

Python

pipeline_name = "anysphere_organization_pipeline"
connection_name = "<anysphere-organization-connection>"
pipeline_spec = {
  "name": pipeline_name,
  "ingestion_definition": {
    "connection_name": connection_name,
    "objects": [
      {
        "table": {
          "source_schema": "default",
          "source_table": "organization_groups",
          "destination_catalog": "main",
          "destination_schema": "anysphere_organization_data",
          "destination_table": "organization_groups"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "organization_group_members",
          "destination_catalog": "main",
          "destination_schema": "anysphere_organization_data",
          "destination_table": "organization_group_members"
        }
      },
      {
        "table": {
          "source_schema": "default",
          "source_table": "organization_members",
          "destination_catalog": "main",
          "destination_schema": "anysphere_organization_data",
          "destination_table": "organization_members"
        }
      }
    ]
  }
}
json_payload = json.dumps(pipeline_spec, indent=2)
create_pipeline(json_payload)

Declarative Automation Bundles のジョブ定義ファイル

Declarative Automation Bundles

YAML

resources:
  jobs:
    anysphere_organization_job:
      name: anysphere_organization_job
      schedule:
        quartz_cron_expression: '0 0 0 * * ?'
        timezone_id: 'UTC'
      tasks:
        - task_key: anysphere_organization_ingestion
          pipeline_task:
            pipeline_id: ${resources.pipelines.anysphere_organization_pipeline.id}

一般的なパターン

高度なパイプライン構成については、「管理対象取り込みパイプラインの一般的なパターン」を参照してください。

次のステップ

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

その他のリソース

• Anysphere 組織コネクタの制限事項
• Anysphere Organization コネクタのリファレンス
• Anysphere Organization コネクタの FAQ