メインコンテンツまでスキップ
に最終更新

カスタムコネクターを構築する

備考

ベータ版

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

カスタムコネクタを使用すると、Lakeflow Connectがマネージドコネクタでサポートしていないソースからデータを取り込むことができます。コネクタを構築およびテストし、独自のDatabricksワークスペースにデプロイしてランします。これを使用するために、コミュニティに登録したり、共有リポジトリにコントリビュートしたりする必要はありません。

GitHub の Lakeflow コミュニティ Connectors リポジトリにあるツールとテンプレートを使用して、コネクタを開発します。このリポジトリには、ソース調査、認証設定、実装、テストなど、各フェーズを支援する AI 搭載の開発ツールが含まれています。これらのツールを使用しても、コネクタがコミュニティコネクタになるわけではありません。リポジトリにはフレームワークとサンプルが用意されており、コントリビュートすることを選択しない限り、コネクタはワークスペース内に保持されます。

後でコネクタを他のユーザーと共有したい場合は、オプションとしてコミュニティにコントリビュートできます。既存のコミュニティコネクタを使用するには、Lakeflow Connectのコミュニティコネクタを参照してください。

要件​

起動する前に、次の項目が揃っていることを確認してください。

  • Python 3.10以降
  • Unity Catalogが有効になっているDatabricksワークスペース
  • 接続先のソースに対するAPI認証情報
  • ローカルにインストールされた Git

リポジトリをセットアップする​

Lakeflow コミュニティ Connectors リポジトリをクローンし、開発用の依存関係をインストールします。

  1. リポジトリのクローンを作成します：

    Bash
    git clone https://github.com/databrickslabs/lakeflow-community-connectors.git
    cd lakeflow-community-connectors

  2. 仮想環境を作成し、依存関係をインストールします：

    Bash
    python -m venv .venv
    source .venv/bin/activate
    pip install -e ".[dev]"

  3. src/databricks/labs/community_connector/sources/ にある既存のコネクタ実装を確認してから、そのパス配下の新しいディレクトリでコネクタの開発を起動します。リポジトリのAI支援による開発コマンドとスキルに従ってください。推奨されるワークフローについては、以下を使用してください。

    Text
    /develop-connector <your-source>
    /validate-connector <your-source>

LakeflowConnectインターフェースを実装します​

各コネクタは LakeflowConnect インターフェースを実装しており、これによってコネクタの認証、テーブルの検出、スキーマの返却、データの読み取り方法が定義されます。

Python
class LakeflowConnect:
    def __init__(self, options: dict[str, str]) -> None:
        """Initialize with connection parameters"""

    def list_tables(self) -> list[str]:
        """Return names of all tables supported by this connector."""

    def get_table_schema(self, table_name: str, table_options: dict[str, str]) -> StructType:
        """Return the Spark schema for a table."""

    def read_table_metadata(self, table_name: str, table_options: dict[str, str]) -> dict:
        """Return metadata: primary_keys, cursor_field, ingestion_type
        (snapshot|cdc|cdc_with_deletes|append)."""

    def read_table(self, table_name: str, start_offset: dict,
                   table_options: dict[str, str]) -> (Iterator[dict], dict):
        """Yield records as JSON dicts and return the next offset
        for incremental reads."""

    def read_table_deletes(self, table_name: str, start_offset: dict,
                           table_options: dict[str, str]) -> (Iterator[dict], dict):
        """Optional: Only required if ingestion_type is 'cdc_with_deletes'."""

メソッドの説明​

手法

説明

__init__

接続パラメーターを辞書として受け取り、ソース用のAPIクライアントを初期化します。

list_tables

コネクタが公開するすべてのテーブル（またはAPI Endpoint）の名前を返します。Databricks は、このリストを使用してテーブル選択 UI を設定します。

get_table_schema

指定されたテーブルのスキーマを記述する Spark StructType を返します。最初のパイプラインランの前、およびスキーマ進化が有効な場合は各ランで呼び出されます。

read_table_metadata

primary_keys、cursor_field、および ingestion_type を含むディクショナリを返します。ingestion_type は、snapshot、cdc、cdc_with_deletes、または append のいずれかである必要があります。

read_table

レコードを Python ディクショナリとして生成し、増分読み取りのための次のオフセットを返します。最初のランでは、start_offset は空です。後続のランでは、前のランによって返されたオフセットが含まれます。

read_table_deletes

オプション。ingestion_type が cdc_with_deletes である場合にのみ、このメソッドを実装してください。削除されたレコードキーを生成し、次のオフセットを返します。

手法

説明

__init__

接続パラメーターを辞書として受け取り、ソース用のAPIクライアントを初期化します。

list_tables

コネクタが公開するすべてのテーブル（またはAPI Endpoint）の名前を返します。Databricks は、このリストを使用してテーブル選択 UI を設定します。

get_table_schema

指定されたテーブルのスキーマを記述する Spark StructType を返します。最初のパイプラインランの前、およびスキーマ進化が有効な場合は各ランで呼び出されます。

read_table_metadata

primary_keys、cursor_field、および ingestion_type を含むディクショナリを返します。ingestion_type は、snapshot、cdc、cdc_with_deletes、または append のいずれかである必要があります。

read_table

レコードを Python ディクショナリとして生成し、増分読み取りのための次のオフセットを返します。最初のランでは、start_offset は空です。後続のランでは、前のランによって返されたオフセットが含まれます。

read_table_deletes

オプション。ingestion_type が cdc_with_deletes である場合にのみ、このメソッドを実装してください。削除されたレコードキーを生成し、次のオフセットを返します。

コネクタの開発​

新しいコネクタを構築および検証するには、次のステップに従います。

  1. ソース API の調査 : ソースの API 仕様、認証メカニズム、レート制限、および利用可能なデータスキーマを調査します。公開するテーブルまたはEndpointを特定します。

  2. 認証の設定 : 接続仕様を生成し、ソースの認証情報を構成し、開発環境からの接続性を検証します。

  3. コネクタの実装 : ソース API に接続し、予期された形式でデータを返すために必要な LakeflowConnect インターフェイスメソッドをすべてコーディングします。

  4. テストと反復 : 実際のソースシステムに対して標準テストスイートをランし、問題を修正します。詳細については、コネクタのテストを参照してください。

  5. コネクタのドキュメント化 : ユーザー向けの README.md を作成し、コネクタの構成可能なパラメーターを記述したコネクタ仕様のYAMLファイルを生成します。

  6. デプロイメントアーティファクトのビルド : ビルドスクリプトをランして、ワークスペースにデプロイ可能な単一ファイルのアーティファクトを作成します。

コネクターをテストする​

このリポジトリには、いくつかのテスト手法が用意されています。

汎用テストスイート（必須）​

提供された認証情報を使用して実際のソースに接続し、認証、スキーマ検出、データ読み取りを含むエンドツーエンドの機能を検証します。

Bash
python -m pytest tests/generic/ --connector <your-source> --credentials credentials.json

ライトバックテスト（推奨）​

書き込み・読み取り・検証サイクルを実行して、増分読み取りと削除を検証します。これにより、オフセット追跡と CDC ロジックが正しく機能していることが確認されます。

Bash
python -m pytest tests/writeback/ --connector <your-source> --credentials credentials.json

単体テスト​

ページネーション処理、型強制、エラー回復など、コネクタ内の複雑なカスタムロジックに対してユニットテストを作成します。

デプロイメントアーティファクトをビルドする​

コネクターがテストスイートに合格したら、Mergeスクリプトをランして、単一ファイルのデプロイ用アーティファクトを生成します。パイプラインは、リポジトリ全体ではなく、ランタイム時にこのファイルを使用します。

Bash
python tools/scripts/merge_python_source.py --connector <your-source>

これにより、すべてのコネクターコードと依存関係を含む、自己完結型の Python ファイルが dist/<your-source>/ に生成されます。

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

独自の Databricks ワークスペースにコネクターをデプロイして実行します：

  1. Databricks ワークスペースのサイドバーで、 +新規 > データを追加またはupload をクリックし、カスタムコネクターを追加するオプションを選択します。

  2. ソース名 には、コネクターの名前を入力します。

  3. 「 GitHubリポジトリURL 」に、コネクタのソースコードをホストしている GitHub リポジトリの URL を入力します。

  4. コネクターを追加 をクリックします。

  5. + Create connection をクリックするか、既存の接続を選択してから、 Next をクリックします。

  6. パイプライン名 に、パイプラインの名前を入力します。

  7. イベントlogの場所 に、カタログ名とスキーマ名を入力します。Databricks は、ここにパイプラインイベントログを保存します。取り込まれたテーブルも、defaultでここに書き込まれます。

  8. 「 ルートパス 」に、ワークスペースパス（例：/Workspace/Users/<your-email>/connectors）を入力します。Databricks は、コネクタのソースコードをここに複製して保存します。

  9. パイプラインの作成 をクリックします。

  10. パイプラインエディターで ingest.py を開き、 objects フィールドを修正して、取り込みたいテーブルを含めます。例えば：

    Python
    from databricks.labs.community_connector.pipeline import ingest

    pipeline_spec = {
        "connection_name": "my_connector_connection",  # Required: UC connection name
        "objects": [
            {"table": {"source_table": "my_table"}},
        ],
    }

    ingest(spark, pipeline_spec)

  11. パイプラインを手動でランするか、スケジュールを設定します。

パイプラインの構成オプション​

ingest.py で次のオプションを構成できます：

オプション

説明

connection_name

必須。ソースの認証資格情報を保存する接続の名前。

objects

必須。取り込むテーブルのリスト。各エントリの形式は {"table": {"source_table": "..."}} です。table オブジェクト内に、オプションで destination_table を指定することもできます。

destination_catalog

取り込まれたテーブルが書き込まれるカタログ。パイプライン作成時に設定されたカタログがdefaultになります。

destination_schema

取り込まれたテーブルが書き込まれるスキーマです。パイプライン作成時に設定されたスキーマがdefaultで使用されます。

scd_type

緩やかに変化するディメンション（Slowly Changing Dimension）戦略: SCD_TYPE_1、SCD_TYPE_2、または APPEND_ONLY。default は SCD_TYPE_1 です。

primary_keys

テーブルのdefaultのプライマリキーを上書きします。列名のリストを指定します。

オプション

説明

connection_name

必須。ソースの認証資格情報を保存する接続の名前。

objects

必須。取り込むテーブルのリスト。各エントリの形式は {"table": {"source_table": "..."}} です。table オブジェクト内に、オプションで destination_table を指定することもできます。

destination_catalog

取り込まれたテーブルが書き込まれるカタログ。パイプライン作成時に設定されたカタログがdefaultになります。

destination_schema

取り込まれたテーブルが書き込まれるスキーマです。パイプライン作成時に設定されたスキーマがdefaultで使用されます。

scd_type

緩やかに変化するディメンション（Slowly Changing Dimension）戦略: SCD_TYPE_1、SCD_TYPE_2、または APPEND_ONLY。default は SCD_TYPE_1 です。

primary_keys

テーブルのdefaultのプライマリキーを上書きします。列名のリストを指定します。

コネクタをコミュニティに提供する​

コネクタをコミュニティに提供することは任意です。コネクタは、それなしでもワークスペース内でランします。他のユーザーがコネクタを見つけて使用できるように共有したい場合は、Lakeflow コミュニティ Connectors リポジトリでプルリクエストを開いてください。提供されたコネクタは コミュニティコネクタ となり、コミュニティによって保守されます。これらは Databricks の SLA の対象外です。

このページの見出し