Build a custom connector for Lakeflow Connect

備考

ベータ版

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

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

GitHub の Lakeflow Community Connectors リポジトリにあるツールとTemplateを使用して、コネクタを開発します。このリポジトリには、ソース調査、認証設定、実装、テストなど、各フェーズを支援するための AI 搭載開発ツールが含まれています。

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

要件

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

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

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

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

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

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

仮想環境を作成し、依存関係をインストールします：
Bash
```
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
```
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'."""

メソッドの説明

次の表は、LakeflowConnect インターフェイスの各メソッドについて説明しています。

手法	説明
`__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` である場合にのみ、このメソッドを実装してください。削除されたレコードキーを生成し、次のオフセットを返します。

コネクタの開発

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

ソース API の調査 : ソースの API 仕様、認証メカニズム、レート制限、および利用可能なデータスキーマを調査します。公開するテーブルまたはEndpointを特定します。
認証の設定 : 接続仕様を生成し、ソースの認証情報を構成し、開発環境からの接続性を検証します。
コネクタの実装 : ソース API に接続し、予期された形式でデータを返すために必要な LakeflowConnect インターフェイスメソッドをすべてコーディングします。
テストと反復 : 実際のソースシステムに対して標準テストスイートをランし、問題を修正します。詳細については、コネクタのテストを参照してください。
コネクタのドキュメント化 : ユーザー向けの README.md を作成し、コネクタの構成可能なパラメーターを記述したコネクタ仕様のYAMLファイルを生成します。
デプロイメントアーティファクトのビルド : ビルドスクリプトをランして、ワークスペースにデプロイ可能な単一ファイルのアーティファクトを作成します。

コネクターをテストする

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

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

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

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

単体テスト

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

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

コネクタがテストスイートに合格したら、パイプラインで実行できるようにパッケージ化します。コネクタは2つのパーツとしてデプロイされます：

単一ファイルのソースアーティファクト。 Mergeスクリプトをランして、コネクタを1つの自己完結型Pythonファイルにフラット化します。パイプラインは、リポジトリ全体ではなく、ランタイム時にこのファイルを使用します。
Bash
```
python tools/scripts/merge_python_source.py --connector <your-source>
```
スクリプトはこのファイルを dist/<your-source>/ に書き込みます。
コネクタの依存関係に必要な Python Wheels (.whl)。 コネクタフレームワークおよびコネクタがインポートするサードパーティライブラリは、Unity Catalog ボリュームに格納された Wheel としてパイプラインから利用可能である必要があります。これらの Wheel が不足していると、ソース検出中にパイプラインが失敗する可能性があります。自分でuploadして UI で参照することも、Community Connector CLI にビルドとuploadを任せることもできます。Community Connector CLI を使用したデプロイを参照してください。

コネクタをデプロイするには、次の2つの方法のいずれかを使用します。

Databricks UI は、ポイント＆クリックによるパスです。 「ライブラリの依存関係」 フィールドでコネクタのwheelを自身で提供する場合、1回限りのデプロイメントに使用します。「Databricks UI でのデプロイ」を参照してください。
community-connector CLI はスクリプト可能なパスです。ローカルで開発していてコネクタのwheelを自動的にビルドおよびuploadしたい場合や、自動化可能な反復可能なデプロイを行いたい場合に使用します。コミュニティコネクタCLIを使用したデプロイを参照してください。

Databricks UI でデプロイする

Databricks UI でのコネクターのデプロイは、コネクターの追加とパイプラインの作成という2つのフェーズで行います。

カスタムコネクターを追加

まず、 データの追加 ページにタイルとして表示されるようにコネクターを追加します。

Databricksワークスペースのサイドバーで、 [+新規] > [データを追加またはupload] をクリックし、 コミュニティコネクタ の下でカスタムコネクタを追加します。
ソース名 には、コネクターの名前を入力します。これは、コネクターのソースコードを含むディレクトリ名（sources/<source-name>）と一致する必要があります。
表示名 には、コネクタのわかりやすい名前を入力します。これを空白のままにすると、defaultでソース名が使用されます。
ライブラリの依存関係 には、コネクターが必要とするPython wheel（.whl）ファイルをUnity Catalogボリュームから追加します。デプロイメントアーティファクトのビルドを参照してください。
接続仕様 には、connector_spec.yamlファイルに対応するYAML形式のコネクターの接続仕様を貼り付けます。
保存をクリックします。コネクタは、 コミュニティコネクタ の下に カスタム タイルとして表示されます。

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

次に、ソースからデータを取り込むパイプラインを作成します。

コネクタタイルを選択して、 データの取り込み ウィザードを開きます。
接続ステップで、 + 接続を作成 をクリックするか既存の接続を選択し、ソースの接続詳細を入力してから、次へをクリックします。
Ingestion setup ステップで、 パイプライン名 を入力し、 Event log location (カタログとスキーマ) を設定して、 コンピュートタイプ を選択してから、 Create pipeline and continue をクリックします。
ソース ステップで、取り込むテーブルを選択します。
配信先 ステップで、取り込まれたテーブルが書き込まれるカタログとスキーマを選択します。
スケジュールと通知 ステップで、オプションのスケジュールと通知を設定し、完了します。
パイプラインを手動で、またはスケジュールに従って実行します。

パイプラインをさらに構成するには、パイプラインエディターで ingest.py を編集します。「パイプライン構成オプション」を参照してください。

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

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のプライマリキーを上書きします。列名のリストを指定します。

Community Connector CLI を使用したデプロイ

CLIのpublishコマンドは、ローカルソースからフレームワークとコネクタのwheelをビルドし、それらをUnity Catalogボリュームにuploadして、コネクタマニフェストにパスを記録し、 データの追加 ページの カスタム タイルとしてコネクタを公開します。アップストリームリポジトリでコネクタが検索されないように、ローカルのコネクタ仕様を指定します。

Bash
community-connector publish <your-source> \
  --spec src/databricks/labs/community_connector/sources/<your-source>/connector_spec.yaml

既にビルド済みのホイールを再利用してビルドステップをスキップするには、--package を使用して渡します。以前に公開したコネクターを置き換えるには、--overwrite を追加します。--package、--volume-path、--catalog、--schema を含むオプションの全リストについては、publish コマンドリファレンスを参照してください。

接続の作成、取り込みパイプラインの作成と更新、公開、公開解除など、ワークフロー全体をコマンドラインから実行するには、Community Connector CLI リファレンスを参照してください。

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

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

要件​

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

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

メソッドの説明​

コネクタの開発​

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

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

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

単体テスト​

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

Databricks UI でデプロイする​

カスタムコネクターを追加​

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

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

Community Connector CLI を使用したデプロイ​

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

要件