Structured Streamingを使ってLakebaseや外部PostgreSQLデータベースに書き込み、組み込みのバッチ処理、自動再試行、ワークスペース管理認証機能を備えています。
Lakebase シンクを使用するタイミング
Lakebaseのシンクは、Lakebaseや外部PostgreSQLデータベースへの低遅延ストリーミング書き込みに使います。 このシンクでは、バッチ処理、接続管理、エラー処理を処理するためにカスタム foreach 関数を実装する必要はありません。
一般的なユース ケースは次のとおりです。
- 運用ダッシュボードまたは顧客向けの機能のために、アプリケーション データベースをリアルタイムで更新します。
- 集計されたストリーミング結果やフィルター処理されたストリーミング結果など、絶えず変化するデータをトランザクション データベースに同期します。
- リアルタイム モードを使用して、2 秒未満の待機時間で構造化ストリーミング クエリの出力を Lakebase テーブルに書き込みます。
Lakebase から Lakehouse の Delta Lake テーブルにデータを同期するには、逆方向の「 Lakebase Change Data Feed」を参照してください。
要件
-
Databricks Runtime 18 LTS 以降。
- 外部PostgreSQL接続および間隔データ型には Databricks Runtime 19 以上が必要です。
- 専用または標準アクセスモードを備えたクラシックコンピュート、またはノートブックやジョブ向けのサーバーレスコンピュートなどがあります。 サーバーレス計算では
Trigger.AvailableNow()を使いましょう。 サーバーレス コンピューティングでのストリーミングを参照してください。 - Lakebase データベース、または外部 PostgreSQL データベースへの Unity Catalog 接続。
識別子の要件
すべてのターゲットに対して、Databricksはスキーマ、テーブル、カラム、プライマリーキーのカラム名(文字またはアンダースコアで始まり、文字、数字、アンダースコアのみを含む)を使用することを推奨しています。 シンクは自動的にLakebaseテーブルを作成することでこれらの要件を強制します。 これらの要件を満たさない識別子を使用するには、クエリを始める前にターゲットテーブルを作成してください。
データベースへの接続
Lakebase シンクでは、次の接続方法がサポートされています。
Unity カタログに登録されている Lakebase テーブル
Unity カタログに登録されている Lakebase テーブルの場合、コネクタは資格情報を自動的に管理し、クエリを実行しているユーザーまたはサービス プリンシパルの ID を使用します。 テーブルが存在しない場合は、コネクタによってテーブルが作成されます。
Lakebase データベースを Unity カタログに登録するには、「 Unity カタログに 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>")
)
Scala
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>: ターゲット テーブルの完全修飾名。catalogは、Lakebase データベースの登録時に作成した Unity カタログです。「Unity カタログに Lakebase データベースを登録する」を参照してください。 テーブルが存在しない場合は、コネクタによって作成されます。 -
<primary-key-columns>: オプション。 ターゲットテーブルの主キーのすべての列をカンマ区切られたリスト(例えばidやuser_id,event_type)。 Upsertの行動を参照してください。 -
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納する Unity カタログ ボリューム パス。 クラウド オブジェクト ストレージ URI を使用することもできます。 場所は、ローカル ディスクではなく書き込み可能なストレージであり、各ストリーミング クエリに固有である必要があります。 これはターゲット テーブルに依存しません。 「構造化ストリーミング チェックポイント」を参照してください。
batchsizeやbatchintervalなどのオプションの構成については、「構成オプション」を参照してください。
Unity カタログに登録されていない Lakebase テーブル
Unity カタログに登録されていない Lakebase テーブルの場合、コネクタは資格情報を自動的に管理し、クエリを実行しているユーザーまたはサービス プリンシパルの 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()
)
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-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データベースの名前。 既定値はdatabricks_postgresです。 データベース の管理を参照してください。 -
<schema>.<table>:schema.table形式のターゲット テーブル。 スキーマを省略すると、シンクはpublicスキーマを使用します。 自動テーブル作成には、文字または下線で始まり、文字、数字、アンダースコアのみを含む識別子を使用してください。 -
<primary-key-columns>: オプション。 ターゲットテーブルの主キーのすべての列をカンマ区切られたリスト(例えばidやuser_id,event_type)。 Upsertの行動を参照してください。 -
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納する Unity カタログ ボリューム パス。 クラウド オブジェクト ストレージ URI を使用することもできます。 場所は、ローカル ディスクではなく書き込み可能なストレージであり、各ストリーミング クエリに固有である必要があります。 これはターゲット テーブルに依存しません。 「構造化ストリーミング チェックポイント」を参照してください。
batchsizeやbatchintervalなどのオプションの構成については、「構成オプション」を参照してください。
Unity Catalog認証情報を用いた外部PostgreSQL
Databricks Runtime 19以降では、コード内に認証情報を保存せずに外部PostgreSQLデータベースに認証するためにUnity Catalog接続を使ってください。 ターゲットテーブルはすでに存在している必要があります。
タイプ POSTGRESQLの接続を作成するには、「 接続を作成する」を参照してください。 クエリを実行するユーザーまたはサービスプリンシパルは接続に 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()
)
Scala
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 カタログ接続の名前です。 -
<database>: ターゲットとなるPostgreSQLデータベースの名前。 -
<schema>.<table>既存のターゲットテーブルをschema.table形式で提供します。 スキーマを省略すると、シンクはpublicスキーマを使用します。 -
<primary-key-columns>: オプション。 ターゲットテーブルの主キーのすべての列をカンマ区切られたリスト(例えばidやuser_id,event_type)。 Upsertの行動を参照してください。 -
/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: クエリがチェックポイントを格納する Unity カタログ ボリューム パス。 クラウド オブジェクト ストレージ URI を使用することもできます。 場所は、ローカル ディスクではなく書き込み可能なストレージであり、各ストリーミング クエリに固有である必要があります。 これはターゲット テーブルに依存しません。 「構造化ストリーミング チェックポイント」を参照してください。
PostgreSQLの接続は常にTLSを使用します。 証明書の検証は、 接続を作成する際に選択するUnity Catalog接続の設定に従って行われます:
-
信頼サーバー証明書:選択すると接続は
sslmode=requireを使用し、サーバー証明書の検証なしに接続を暗号化します。 -
ユーザー提供のサーバー証明書:Trust server証明書を選択していない場合
sslmode=verify-full使用するためにPEMでエンコードされたサーバー証明書を提供します。 証明書を提供しなければ、接続はJVMのデフォルトトラストストアでsslmode=verify-fullを使用します。
構成オプション
シンクでは、認識できないオプション ( 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")。
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は省略可能です。 もしそれを省略してブランチに単一の読み書きエンドポイントがある場合、シンクはデフォルトでそのエンドポイントを選択します。 |
Unity Catalog認証情報を用いた外部PostgreSQL
Unity Catalogの認証情報で外部PostgreSQLデータベースに接続した場合、以下のオプションが適用されます。
| Key | Default | 説明 |
|---|---|---|
database |
None | Required. ターゲット PostgreSQL データベース名。 |
databricks.connection |
None | Required. Unity Catalog の接続名は、Unity Catalog 管理による外部 PostgreSQL への認証です。 |
dbtable |
None | Required. 既存のターゲットテーブル名を schema.table 形式で使います。 スキーマを指定しない場合、既定のスキーマ値は public。 |
データ型マッピング
シンクは、各DataFrame列が対応するターゲット列と互換性があるかを確認し、既存のLakebaseや外部PostgreSQLテーブルに書き込みます。
以下の表は、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 |
以下の表は、Databricks Runtime 19以上でサポートされている型を示しています:
| Spark の種類 | 自動生成のLakebaseテーブルタイプ | 既存のPostgreSQLテーブルにおける互換性型 |
|---|---|---|
DayTimeIntervalType、YearMonthIntervalType |
interval |
interval |
アップサートの挙動
upsertkeyオプションはターゲットテーブルの主キー列を識別します。 既存のテーブルの場合、 upsertkey の列はテーブルの主キーと正確に一致しなければなりません。 オプションを省略すると、シンクはテーブルから主キーを読み取ります。 シンクが作成するLakebaseテーブルの場合、 upsertkey がプライマリキーを定義します。 オプションを省略すると、シンクは主キーのないテーブルを作成します。
ターゲットテーブルに主キーがある場合、シンクはPostgreSQLの文法でアップサート INSERT INTO ... ON CONFLICT (<primary_key_columns>) DO UPDATE SET ... します。 ターゲットテーブルにプライマリキーがない場合、シンクはインサートを実行します。 クエリの出力モードはこの挙動に影響を与えません。
すべての主キー列はDataFrameに存在し、数値型や文字列型などの比較可能な型を使用する必要があります。
パフォーマンスチューニング
バッチ処理とバックプレッシャ
いずれかの条件が満たされると、フラッシュがトリガーされます。
- バッファーは
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
この表は、クラシックおよびサーバーレスコンピュートにおける構造化ストリーミングトリガータイプのサポートを示しています:
| トリガー | クラシック コンピューティング | サーバーレスコンピュート(ノートブックとジョブ) |
|---|---|---|
RealTime |
Yes | No |
ProcessingTime |
Yes | No |
AvailableNow |
Yes | Yes |
Once |
Yes. Deprecated.
AvailableNow を使用してください。 |
Yes. Deprecated.
AvailableNow を使用してください。 |
出力モード
次の表は、構造化ストリーミング出力モードのサポートを示しています。
| 出力モード | Supported |
|---|---|
update |
Yes |
append |
Yes. 動作は、 updateと同じです。 ターゲット テーブルに主キーがある場合はクエリがアップサートされ、それ以外の場合はクエリが挿入されます。
Upsertの行動を参照してください。 |
complete |
No |
制限事項
- Unityカタログ接続を介して接続された外部PostgreSQLデータベースの場合、ターゲットテーブルはすでに存在している必要があります。 シンクはLakebaseでのみ自動的に欠落テーブルを生成します。
- レイクフローパイプラインはサポートされていません。