Lakebase に接続する

重要

この機能は パブリック プレビュー段階です。

構造化ストリーミングを使用して、組み込みのバッチ処理、自動再試行、ワークスペース管理認証を使用して Lakebase に書き込みます。

Lakebase シンクを使用するタイミング

Lakebase への低待機時間のストリーミング書き込みには、Lakebase シンクを使用します。 このシンクでは、バッチ処理、接続管理、エラー処理を処理するためにカスタム foreachBatch 関数を実装する必要はありません。

一般的なユース ケースは次のとおりです。

  • 運用ダッシュボードまたは顧客向けの機能のために、アプリケーション データベースをリアルタイムで更新します。
  • 集計されたストリーミング結果やフィルター処理されたストリーミング結果など、絶えず変化するデータをトランザクション データベースに同期します。
  • リアルタイム モードを使用して、2 秒未満の待機時間で構造化ストリーミング クエリの出力を Lakebase テーブルに書き込みます。

Lakebase から Lakehouse の Delta Lake テーブルにデータを同期するには、逆方向の「 Lakebase Change Data Feed」を参照してください。

要件

  • Databricks Runtime 18 以降
  • 専用または標準のアクセス モードを使用したクラシック コンピューティング。
  • Lakebase データベース

データベースへの接続

Lakebase シンクでは、次の接続方法がサポートされています。

Unity カタログに登録されている Lakebase テーブル

Unity カタログに登録されている Lakebase テーブルの場合、コネクタは資格情報を自動的に管理し、クエリを実行しているユーザーまたはサービス プリンシパルの ID を使用します。 テーブルが存在しない場合は、コネクタによってテーブルが作成されます。

Lakebase データベースを Unity カタログに登録するには、「 Unity カタログに Lakebase データベースを登録する」を参照してください。

Lakebase テーブルに書き込むには、完全修飾テーブル名 .toTable() を指定して、catalog.schema.table メソッドを使用します。 次の例は、必要なオプションとオプションの upsertkey オプションを示しています。

Python

(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

Scala

df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")

次のプレースホルダーを置き換えてください。

  • <catalog>.<schema>.<table>: ターゲット テーブルの完全修飾名。 catalogは、Lakebase データベースの登録時に作成した Unity カタログです。「Unity カタログに Lakebase データベースを登録する」を参照してください。 テーブルが存在しない場合は、コネクタによって作成されます。
  • <primary-key-column>: オプション。 アップサート キーを形成する列のコンマ区切りのリスト ( id や user_id,event_typeなど)。 upsertkeyを省略すると、シンクはターゲット テーブルの主キーからキーを推論します。 Upsert の動作を参照してください。
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納する Unity カタログ ボリューム パス。 クラウド オブジェクト ストレージ URI を使用することもできます。 場所は、ローカル ディスクではなく書き込み可能なストレージであり、各ストリーミング クエリに固有である必要があります。 これはターゲット テーブルに依存しません。 「構造化ストリーミング チェックポイント」を参照してください。

batchsizeやbatchintervalなどのオプションの構成については、「構成オプション」を参照してください。

Unity カタログに登録されていない Lakebase テーブル

Unity カタログに登録されていない Lakebase テーブルの場合、コネクタは資格情報を自動的に管理し、クエリを実行しているユーザーまたはサービス プリンシパルの ID を使用します。 テーブルが存在しない場合は、コネクタによってテーブルが作成されます。

Lakebase テーブルに書き込むには、 endpoint オプションと dbtable オプションを使用します。 次の例には、省略可能な database オプションと upsertkey オプションも含まれています。

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  # Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  // Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

次のプレースホルダーを置き換えてください。

  • <project-id>.<branch-id>.<endpoint-id>: あなたの Lakebase のエンドポイント。 [コンピューティング] タブの [ID の取得] メニューの [リソース名] で 3 つの値をすべて検索します。このメニューの形式はprojects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>。 「 コンピューティング識別子」を参照してください。
  • <database>: オプション。 ターゲット Postgres データベースの名前。 既定値は databricks_postgres です。 データベース の管理を参照してください。
  • <schema>.<table>: schema.table 形式のターゲット テーブル。 スキーマを省略すると、シンクは public スキーマを使用します。 文字またはアンダースコアで始まり、文字、数字、およびアンダースコアのみを含む単純な識別子を使用します。引用符で囲まれた識別子と特殊文字 (ハイフンなど) はサポートされていません。
  • <primary-key-column>: オプション。 アップサート キーを形成する列のコンマ区切りのリスト ( id や user_id,event_typeなど)。 upsertkeyを省略すると、シンクはターゲット テーブルの主キーからキーを推論します。 Upsert の動作を参照してください。
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納する Unity カタログ ボリューム パス。 クラウド オブジェクト ストレージ URI を使用することもできます。 場所は、ローカル ディスクではなく書き込み可能なストレージであり、各ストリーミング クエリに固有である必要があります。 これはターゲット テーブルに依存しません。 「構造化ストリーミング チェックポイント」を参照してください。

batchsizeやbatchintervalなどのオプションの構成については、「構成オプション」を参照してください。

構成オプション

シンクでは、認識できないオプション ( JDBC_STREAMING_SINK_INVALID_OPTIONS) のエラーが発生します。

次のオプションは、すべての接続方法に適用されます。

Key Default 説明
batchinterval 100 milliseconds Optional. フラッシュ前にバッファー内の行を保持する最大時間。 たとえば、「 "50 milliseconds" 」のように入力します。
batchsize 1000 Optional. 各データベース トランザクションの最大行数。
checkpointLocation None Required. Unity カタログ ボリューム (/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>) などのチェックポイント ディレクトリへのパス。 各クエリに対して一意である必要があります。 「構造化ストリーミング チェックポイント」を参照してください。
upsertkey None Optional. アップサート キーを構成する列名のコンマ区切りリスト。 たとえば、"id" または "user_id,event_type" です。 upsertkeyを指定した場合、列はテーブルの主キーと一致する必要があります。そうしないと、クエリが失敗します。 これを省略すると、シンクは主キーを自動的に使用します。 詳細については、「 Upsert の動作」を参照してください。

Unity カタログに登録されていない Lakebase テーブル

Unity カタログに登録されていない Lakebase テーブルに接続する場合は、次のオプションが適用されます。

Key Default 説明
database databricks_postgres Optional. ターゲット PostgreSQL データベース名。
dbtable None Required. schema.table形式のターゲット テーブル名。 スキーマを指定しない場合、既定のスキーマ値は public。 文字またはアンダースコアで始まり、文字、数字、およびアンダースコアのみを含む単純な識別子を使用します。 テーブル名またはスキーマ名を引用符で囲む必要はありません。引用符で囲まれた識別子と、ハイフンなどの特殊文字を含む名前はサポートされていません。
endpoint None Required. project_id.branch_idまたはproject_id.branch_id.endpoint_id形式の Lakebase エンドポイント。 endpoint_idは省略可能です。省略すると、ブランチに単一の読み取り/書き込みエンドポイントがある場合、シンクは既定でそのエンドポイントを選択します。

Upsert の動作

アップサート キーが存在する場合は、 upsertkey で指定するか、テーブルの主キーからシンクによって推論されると、シンクは PostgreSQL の INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ... 構文を使用してテーブルにアップサートします。

アップサート キーが存在しない場合、シンクは挿入を実行します。 クエリの出力モードは、アップサートまたは挿入の動作には影響しません。

upsertkey列は次の必要があります。

  • DataFrame 列の空でないサブセットにする。
  • ターゲット テーブルの PRIMARY KEY と正確に一致します。 指定した列が主キーと一致しない場合、クエリは失敗します。
  • 数値型や文字列型など、同等の型にする。 同時書き込み中にデータベースのデッドロックを防ぐために、シンクは各バッチ内で行を upsert キーで並べ替えます。 Upsert キーは、複合型または構造体型をサポートしていません。

列名は PostgreSQL の既定の二重引用符 "で自動的に引用符で囲まれ、予約済みのキーワードと大文字と小文字が混在する名前が処理されます。

テーブル名とスキーマ名では、文字またはアンダースコアで始まり、文字、数字、アンダースコアのみを含む単純な識別子を使用する必要があります。 シンクは、引用符で囲まれた識別子や特殊文字 (ハイフンなど) をテーブル名またはスキーマ名でサポートしていません。

パフォーマンスチューニング

バッチ処理とバックプレッシャ

いずれかの条件が満たされると、フラッシュがトリガーされます。

  • バッファーは batchsize 行に達し、既定では 1000 に達します。
  • バッファーの有効期間が batchinterval を超えています。既定では 100 milliseconds。

データベースが受信データ レートに対応できない場合、シンクはバックプレッシャをソースにアップストリームに伝達します。

待機時間とスループットのガイダンス:

  • リアルタイム モードの待機時間が短いワークロードの場合は、フラッシュ前の最大時間を短くするために batchinterval を減らします。 コード例の概念とリアルタイム モードの例については、 構造化ストリーミング の リアルタイム モード を参照してください。
  • 高スループットのワークロードの場合は、各トランザクションのオーバーヘッドを減らすために batchsize を増やします。

接続の動作

シンクは Executor 上で接続プーリングを使用します。 既定では、各タスクは 1 つのデータベース接続を使用します。

Databricks では、接続ごとに既定値の 1 タスクを使用することをお勧めします。 接続ごとにタスクの数を増やすと、接続の競合が発生し、高スループット接続の待機時間が長くなる可能性があります。

接続に対するタスクの比率を構成するには、 spark.databricks.sql.streaming.jdbc.tasksPerConnection Spark 構成を設定します。 ターゲット データベースの接続制限が低い場合は、シャッフル パーティションの数を減らすか、 spark.databricks.sql.streaming.jdbc.tasksPerConnectionを増やします。

シンクは、接続エラー、デッドロック、レート制限など、一時的な JDBC エラーを自動的に再試行します。 シンクがすべての再試行を使い果たした場合、クエリは失敗します。

サポートされているトリガーと出力モード

Triggers

次の表は、構造化ストリーミング トリガーの種類のサポートを示しています。

トリガー Supported
realTime Yes
ProcessingTime Yes
AvailableNow Yes
Once Yes

出力モード

次の表は、構造化ストリーミング出力モードのサポートを示しています。

出力モード Supported
update Yes
append Yes. 動作は、 updateと同じです。 ターゲット テーブルに主キーがある場合はクエリがアップサートされ、それ以外の場合はクエリが挿入されます。 Upsert の動作を参照してください。
complete No

制限事項

  • サーバーレス コンピューティングと Lakeflow Spark 宣言パイプラインはサポートされていません。
  • 書き込みターゲットとしてサポートされるのは Lakebase のみです。 外部 PostgreSQL と互換性のあるデータベースはサポートされていません。
  • Last updated on