Lakeflow Connect用のカスタムコネクターを作成する

カスタムコネクターは、Lakeflow Connectがサポートしていないソースからマネージドコネクターでデータを取り込むことができます。 コネクターをビルド・テストし、自社のAzure Databricksワークスペースでデプロイ・実行します。 コミュニティに登録したり、共有リポジトリに寄稿したりする必要はなく、使用可能です。

GitHubのLakeflow Community Connectorsリポジトリにあるツールとテンプレートを使ってコネクターを開発しましょう。 リポジトリには、ソース調査、認証のセットアップ、実装、テストなど、各フェーズを支援する AI を利用した開発ツールが含まれています。

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

必要条件

開始する前に、以下を用意してください。

• Python 3.10 以降
• Unity カタログが有効になっているAzure Databricks ワークスペース
• 接続するソースの API 資格情報
• ローカルにインストールされた Git

リポジトリを設定する

Lakeflow Community Connectors リポジトリを複製し、開発の依存関係をインストールします。

1. リポジトリを複製します。

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

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

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

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

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

LakeflowConnect インターフェイスを実装する

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

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 インターフェース内の各メソッドを説明しています。

コネクタを開発する

新しいコネクタをビルドして検証するには、次の手順に従います。

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

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

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

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

5. コネクタを文書化する: ユーザー向けの README.md を記述し、コネクタの構成可能なパラメーターを記述するコネクタ スペック YAML ファイルを生成します。

6. デプロイ成果物をビルドする: ビルド スクリプトを実行して、ワークスペースにデプロイできる単一ファイルの成果物を生成します。

コネクタをテストする

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

汎用テスト一式 (必須)

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

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

ライトバック テスト (推奨)

書き戻しテストでは、増分読み取りおよび削除を検証するために、書き込み・読み取り・検証のサイクルを実行します。 これにより、オフセット追跡と CDC ロジックが正しく動作することを確認できます。

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

単体テスト

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

デプロイ成果物をビルドする

コネクターがテストスイートに合格したら、パイプラインで実行できるようにパッケージ化してください。 コネクタは2つの部分に分かれて展開されます:

• 単一ファイルのソースアーティファクト。 マージスクリプトを実行してコネクターを一つの独立したPythonファイルにフラット化します。 パイプラインは、完全なリポジトリではなく、実行時にこのファイルを使用します。

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

  スクリプトはこのファイルを dist/<your-source>/に書き込みます。

• コネクタの依存関係を割り当てるためにPythonホイール(.whl)。 コネクタフレームワークおよびコネクタがインポートするサードパーティのライブラリは、Unityカタログボリュームに保存されたホイールとしてパイプラインに利用可能でなければなりません。 これらのホイールが欠けている場合、ソース発見時にパイプラインが故障する可能性があります。 自分でアップロードしてUIで参照することもできますし、Community ConnectorのCLIにビルドしてアップロードしてもらうこともできます。 詳細は「 Community Connector CLIでのデプロイ」を参照してください。

コネクターの展開方法は2通りのいずれかです:

• Azure Databricks の UI は、ポイント アンド クリックで操作する方法です。 ライブラリ 依存関係欄 でコネクタのホイールを自分で提供する際に一度限りのデプロイに使うと良いでしょう。 Databricks UIの「デプロイ」を参照してください。
• community-connectorCLIはスクリプト可能なパスです。 ローカル開発でコネクターのホイールを組み立ててアップロードしてもらいたい時や、自動化可能な繰り返し展開をしたい時に使うと良いでしょう。 詳細は「 Community Connector CLIでのデプロイ」を参照してください。

Databricks UIでデプロイ

Azure Databricks UIでコネクターを2段階展開します:コネクタを追加し、パイプラインを作成します。

カスタムコネクタを追加してください

まず、コネクターを追加して「 データ追加 」ページにタイルとして表示します:

1. Azure Databricksワークスペースのサイドバーで、+New>Add or upload data をクリックし、次に Community connectors でカスタム コネクタを追加します。
2. [ ソース名] に、コネクタの名前を入力します。 これはコネクタのソースコード(sources/<source-name>)を含むディレクトリ名と一致しなければなりません。
3. 表示名にはコネクターのフレンドリーな名前を入力してください。 これを空欄にすると、ソース名がデフォルトになります。
4. ライブラリ依存関係については、コネクターが必要なUnityカタログボリュームから必要なPythonホイール(.whl)ファイルを追加してください。 詳細は 「デプロイメントアーティファクトの作成」を参照してください。
5. 接続仕様については、コネクタの接続仕様をYAMLに貼り付け、connector_spec.yamlファイルと一致させます。
6. 保存 をクリックします。 コネクターはコミュニティコネクターのカスタムタイルとして表示されます。

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

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

1. コネクタータイルを選択して 「データを取り込み 」ウィザードを開きます。
2. 接続ステップで+接続を作成するか、既存の接続を選択し、元の接続情報を入力してから「次へ」をクリックします。
3. 取り込みの設定ステップで、パイプライン名を入力し、イベントログの場所(カタログとスキーマ)を設定し、Computeタイプを選択し、「パイプラインを作成」をクリックして続けます。
4. Sourceステップで取り込むテーブルを選択します。
5. 宛先ステップで、取り込んだテーブルが書かれるカタログとスキーマを選択します。
6. スケジュール と通知 のステップで、任意のスケジュールと通知を設定してから終了します。
7. パイプラインは手動またはスケジュール通りに実行してください。

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

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

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

Community Connector CLIで展開

CLIのpublishコマンドは、ローカルソースからフレームワークとコネクタホイールを構築し、Unityカタログのボリュームにアップロードし、それらのパスをコネクタマニフェストに記録し、Add dataページのカスタムタイルとしてコネクターを公開します。 ローカルのコネクタ仕様を参照するように設定して、アップストリーム リポジトリでコネクタを探さないようにしてください:

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 Community Connectors リポジトリでプルリクエストを開いてください。 貢献したコネクターは コミュニティコネクターとなり、コミュニティが管理し、DatabricksのSLAに縛られていません。