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

Lakebase に接続

Structured Streamingを使用して、組み込みのバッチ処理、自動再試行、ワークスペース管理型認証を備えた Lakebase または外部の PostgreSQL データベースに書き込みます。

Lakebase シンクはいつ使用しますか？​

Lakebase シンクを使用して、Lakebase または外部の PostgreSQL データベースへの低レイテンシのストリーミング書き込みを行います。このシンクでは、バッチ処理、接続管理、エラー処理を処理するためにカスタムの foreach 関数を実装する必要はありません。


ユースケースの例：

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

Lakebase からレイクハウス内のDelta Lakeテーブルにデータを逆方向に同期するには、「Lakebase チェンジデータフィード」を参照してください。

要件​

  • Databricks Runtime 18 LTS 以降。

    • 外部 PostgreSQL 接続を使用するには、Databricks Runtime 19 以降を使用し、 Custom JDBC on UC Compute のプレビューにオプトインする必要があります。
    • インターバル データ型を使用するには、Databricks Runtime 19 以降を使用する必要があります。

  • 専用または標準アクセスモードのクラシックコンピュート、あるいはノートブックやジョブ向けのServerlessコンピュート。サーバーレスコンピュート上で、Trigger.AvailableNow() を使用します。Serverlessコンピュートでのストリーミングを参照してください。

  • Lakebase データベース、または外部 PostgreSQL データベースへの Unity Catalog 接続。

Identifier requirements​

すべてのターゲットについて、Databricks では、文字またはアンダースコアで始まり、文字、数字、およびアンダースコアのみを含むスキーマ、テーブル、列、および主キー列の名前を使用することをお勧めします。シンクは、Lakebase テーブルを自動的に作成するときに、これらの要件を強制します。これらの要件を満たさない識別子を使用するには、クエリーを開始する前にターゲットテーブルを作成してください。

データベースに接続します​

Lakebase sink は次の接続メソッドをサポートしています。

Unity Catalog に登録されている Lakebase テーブル​

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

Lakebase データベースを Unity Catalog に登録するには、Unity Catalog で Lakebase データベースを登録するを参照してください。

Lakebase テーブルに書き込むには、完全修飾テーブル名 catalog.schema.table とともに .toTable() メソッドを使用します。

Python
(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-columns>")  # Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

次のプレースホルダーを置き換えます。

  • <catalog>.<schema>.<table>：ターゲットテーブルの完全修飾名。The catalog is the Unity Catalog catalog you created when you registered the Lakebase database, see Unity CatalogにLakebaseデータベースを登録する.テーブルが存在しない場合は、コネクタが作成します。
  • <primary-key-columns>: 省略可能です。ターゲットテーブルの主キーに含まれるすべての列のコンマ区切りリスト（例: id や user_id,event_type）。アップサートの動作を参照してください。
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納するUnity Catalogボリュームパス。クラウドオブジェクトストレージURIを使用することもできます。ロケーションは、ローカルディスクではなく書き込み可能なストレージである必要があり、各ストリーミングクエリに対して一意である必要があります。これはターゲットテーブルとは無関係です。構造化ストリーミングのチェックポイントを参照してください。

batchsize や batchinterval などのオプショナルな設定については、PostgreSQL シンクオプションを参照してください。

Unity Catalogに登録されていないLakebaseテーブル​

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

Lakebase テーブルに書き込むには、endpoint および dbtable オプションを使用します。

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-columns>")  # Optional
  .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>: オプション。ターゲットのPostgreSQLデータベースの名前。default は databricks_postgres です。詳細については、データベースの管理を参照してください。
  • <schema>.<table>：schema.table 形式のターゲットテーブル。スキーマを省略した場合、シンクは public スキーマを使用します。自動テーブル作成の場合は、文字またはアンダースコアで始まり、文字、数字、アンダースコアのみを含む識別子を使用します。
  • <primary-key-columns>: 省略可能です。ターゲットテーブルの主キーに含まれるすべての列のコンマ区切りリスト（例: id や user_id,event_type）。アップサートの動作を参照してください。
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納するUnity Catalogボリュームパス。クラウドオブジェクトストレージURIを使用することもできます。ロケーションは、ローカルディスクではなく書き込み可能なストレージである必要があり、各ストリーミングクエリに対して一意である必要があります。これはターゲットテーブルとは無関係です。構造化ストリーミングのチェックポイントを参照してください。

batchsize や batchinterval などのオプショナルな設定については、PostgreSQL シンクオプションを参照してください。

Unity Catalog 資格情報を使用する外部 PostgreSQL​

備考

プレビュー

この機能は パブリック プレビュー段階です。ワークスペース管理者は、 Previews ページから Custom JDBC on UC コンピュート へのアクセスを制御できます。Databricksのプレビューを管理するを参照してください。

Unity Catalog 接続を使用すると、コードに資格情報を保存することなく、外部の PostgreSQL データベースに対して認証を行うことができます。ターゲット テーブルがすでに存在している必要があります。

タイプ POSTGRESQL の接続を作成します。接続の作成を参照してください。クエリーを実行するユーザーまたは Databricks Service Principal は、接続に対して USE CONNECTION を持っている必要があります。

PostgreSQLテーブルに書き込むには、databricks.connection、database、および dbtable オプションを使用します。

Python
(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("databricks.connection", "<connection-name>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-columns>")  # Optional
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

次のプレースホルダーを置き換えます。

  • <connection-name>: Unity Catalog接続の名前。
  • <database>：ターゲットの PostgreSQL データベースの名前。
  • <schema>.<table>：schema.table 形式の既存のターゲットテーブル。スキーマを省略した場合、シンクは public スキーマを使用します。
  • <primary-key-columns>: 省略可能です。ターゲットテーブルの主キーに含まれるすべての列のコンマ区切りリスト（例: id や user_id,event_type）。アップサートの動作を参照してください。
  • /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納するUnity Catalogボリュームパス。クラウドオブジェクトストレージURIを使用することもできます。ロケーションは、ローカルディスクではなく書き込み可能なストレージである必要があり、各ストリーミングクエリに対して一意である必要があります。これはターゲットテーブルとは無関係です。構造化ストリーミングのチェックポイントを参照してください。

PostgreSQL接続では常にTLSが使用されます。証明書の検証は、接続を作成する際に選択するUnity Catalog接続の設定に従います。

  • サーバ証明書を信頼する : 選択すると、接続では sslmode=require が使用され、サーバ証明書を検証せずに接続が暗号化されます。
  • ユーザー提供のサーバー証明書 : 「サーバー証明書を信頼する」 が選択されていないときに sslmode=verify-full を使用するには、PEM エンコードされたサーバー証明書を提供します。証明書を指定しない場合、接続では JVM の default トラストストアとともに sslmode=verify-full が使用されます。

構成オプション​

シンクは認識できないオプション、JDBC_STREAMING_SINK_INVALID_OPTIONSに対してエラーを返します。

共通のオプションや各接続方法のオプションなど、シンクの構成オプションについては、PostgreSQL シンクのオプションを参照してください。

データ型のマッピング​

シンクは、既存の Lakebase または外部の PostgreSQL テーブルに書き込む前に、各 DataFrame カラムが対応するターゲットカラムと互換性があることを確認します。

次の表には、Databricks Runtime 18 LTS 以降でサポートされている型が含まれています。

Spark 型

自動的に作成される Lakebase テーブルのタイプ

既存の PostgreSQL テーブル内の互換性のある型

ByteType, ShortType

smallint

smallint

IntegerType

integer

integer

LongType

bigint

bigint

FloatType

real

real

DoubleType

double precision

double precision

DecimalType

numeric

numeric

StringType

text

varchar, text

VarcharType(n)

varchar(n)

varchar, text

CharType(n)

char(n)

char

BinaryType

bytea

bytea

BooleanType

boolean

boolean

TimestampType

timestamptz

timestamptz

TimestampNTZType

timestamp

timestamp

DateType

date

date

ArrayType、MapType、StructType、VariantType、 NullType

jsonb

json, jsonb

Spark 型

自動的に作成される Lakebase テーブルのタイプ

既存の PostgreSQL テーブル内の互換性のある型

ByteType, ShortType

smallint

smallint

IntegerType

integer

integer

LongType

bigint

bigint

FloatType

real

real

DoubleType

double precision

double precision

DecimalType

numeric

numeric

StringType

text

varchar, text

VarcharType(n)

varchar(n)

varchar, text

CharType(n)

char(n)

char

BinaryType

bytea

bytea

BooleanType

boolean

boolean

TimestampType

timestamptz

timestamptz

TimestampNTZType

timestamp

timestamp

DateType

date

date

ArrayType、MapType、StructType、VariantType、 NullType

jsonb

json, jsonb

次の表は、Databricks Runtime 19以降でサポートされているタイプを示しています。

Spark 型

自動的に作成される Lakebase テーブルのタイプ

既存の PostgreSQL テーブル内の互換性のある型

DayTimeIntervalType, YearMonthIntervalType

interval

interval

Spark 型

自動的に作成される Lakebase テーブルのタイプ

既存の PostgreSQL テーブル内の互換性のある型

DayTimeIntervalType, YearMonthIntervalType

interval

interval

Upsert behavior​

upsertkey オプションは、ターゲットテーブルの主キー列を特定します。既存のテーブルの場合、upsertkey 内の列はテーブルの主キーと完全に一致する必要があります。オプションを省略した場合、シンクはテーブルから主キーを読み込みます。シンクが作成する Lakebase テーブルの場合、upsertkey が主キーを定義します。オプションを省略した場合、シンクは主キーなしでテーブルを作成します。

ターゲットテーブルに主キーがある場合、シンクは PostgreSQL の INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ... 構文を使用してアップサートを実行します。ターゲットテーブルに主キーがない場合、シンクは挿入を実行します。クエリーの出力モードはこの動作に影響しません。

すべての主キー列が DataFrame に存在し、数値型や文字列型などの比較可能な型を使用している必要があります。

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

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

フラッシュはいずれかの条件が満たされた場合にトリガーされます。

  • バッファーが batchsize 行に達すると、デフォルトで 1000 に設定されます。
  • バッファー期間がbatchintervalを超えています。デフォルトは100 millisecondsです。

データベースが流入するデータ速度に追いつけなくなると、シンクはアップストリームでバックプレッシャーをソースに伝播します。

レイテンシーとスループットのガイダンス：

  • リアルタイムモードを使用する低レイテンシーのワークロードでは、フラッシュまでの最大時間を短縮するために batchinterval を減らしてください。概念については リアルタイムモードの概念を、コード例については リアルタイムモードの例を参照してください。
  • ハイスループットのワークロードでは、batchsizeを増加させて、各トランザクションのオーバーヘッドを削減します。

接続動作​

シンクはエグゼキューターで接続プールを使用します。デフォルトでは、各タスクは1つのデータベース接続を使用します。

Databricks は、接続ごとに 1 タスクのデフォルト値を使用することをお勧めします。各接続のタスク数を増やすと、接続の競合を発生させ、高スループット接続の遅延が増加する可能性があります。

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

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

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

トリガー​

この表は、クラシックおよび Serverless コンピュートにおける Structured Streaming の Trigger 型のサポートを示しています。

トリガー

Classic Compute

Serverless コンピュート (ノートブックとジョブ)

RealTime

はい

No

ProcessingTime

はい

No

AvailableNow

はい

はい

Once

はい。非推奨です。AvailableNow を使用してください。

はい。非推奨です。AvailableNow を使用してください。

トリガー

Classic Compute

Serverless コンピュート (ノートブックとジョブ)

RealTime

はい

No

ProcessingTime

はい

No

AvailableNow

はい

はい

Once

はい。非推奨です。AvailableNow を使用してください。

はい。非推奨です。AvailableNow を使用してください。

出力モード​

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

出力モード

サポートされています

update

はい

append

はい。動作は update と同一です。ターゲットテーブルに主キーがある場合、クエリーはアップサートを行い、そうでない場合は挿入を行います。アップサートの動作を参照してください。

complete

No

出力モード

サポートされています

update

はい

append

はい。動作は update と同一です。ターゲットテーブルに主キーがある場合、クエリーはアップサートを行い、そうでない場合は挿入を行います。アップサートの動作を参照してください。

complete

No

制限事項​

  • Unity Catalog 接続を通じて接続された外部 PostgreSQL データベースの場合、ターゲットテーブルがすでに存在している必要があります。シンクは、Lakebase 内でのみ不足しているテーブルを自動的に作成します。
  • LakeFlow Pipelines はサポートされていません。
このページの見出し