Kafka コネクタ リファレンス

このページには、Lakeflow Connect のマネージド Kafka コネクタのリファレンス ドキュメントが含まれています。

コネクターオプション

次のオプションは、取り込みパイプラインの各宛先テーブルに対してKafkaソースを構成します。パイプライン定義の connector_options.kafka_options で、これらのオプションを指定してください。完全なパイプラインの例については、「例」を参照してください。

オプション	Type	デフォルト	説明
topics	文字列のリスト：	—	サブスクライブするトピック名のリストです。topic_patternと相互に排他的です。topicsまたはtopic_patternのいずれかが必要です。
topic_pattern	String	—	サブスクライブするトピック名に一致する Java 正規表現です。topicsと相互に排他的です。
starting_offset	String	latest	チェックポイントが存在しない場合に読み取りを開始する場所（初回実行のみ）。有効な値: latest、earliest。
key_transformer	Transformer	—	メッセージキーのデシリアライザ構成。設定されていない場合、キー列は BINARY として保持されます。Transformer オプションを参照してください。
value_transformer	Transformer	—	メッセージ値の逆シリアライザー設定。設定されていない場合、値の列はBINARYとして保持されます。Transformerオプションを参照してください。

オプション	Type	デフォルト	説明
topics	文字列のリスト：	—	サブスクライブするトピック名のリストです。topic_patternと相互に排他的です。topicsまたはtopic_patternのいずれかが必要です。
topic_pattern	String	—	サブスクライブするトピック名に一致する Java 正規表現です。topicsと相互に排他的です。
starting_offset	String	latest	チェックポイントが存在しない場合に読み取りを開始する場所（初回実行のみ）。有効な値: latest、earliest。
key_transformer	Transformer	—	メッセージキーのデシリアライザ構成。設定されていない場合、キー列は BINARY として保持されます。Transformer オプションを参照してください。
value_transformer	Transformer	—	メッセージ値の逆シリアライザー設定。設定されていない場合、値の列はBINARYとして保持されます。Transformerオプションを参照してください。

Transformer オプション

トランスフォーマーは、バイナリのKafkaメッセージのキーと値が構造化された列にどのように逆シリアル化されるかを定義します。シリアル化形式と、key_transformerまたはvalue_transformerの下の対応する形式固有のオプションを指定します。キー、値、またはその両方に対してトランスフォーマーを個別に構成できます。トランスフォーマーが設定されていない場合、列はBINARYとして保持されます。

JSON の場合、明示的なスキーマを提供するか、進化を含むスキーマ推論を使用するか、json_optionsを完全に省略して値をVARIANT列として保存できます。

オプション	適用対象	Type	デフォルト	説明
format	すべて	String	—	データのシリアル化形式。有効な値: STRING、JSON。STRINGには追加オプションは必要ありません。トランスフォーマーにjson_optionsが指定されていない場合、値はデフォルトでVariantとして解析されます。詳細については、Variant Data Formatを参照してください。
json_options.schema	JSON	String	—	Spark DDL形式のインラインスキーマ（例: "id BIGINT, name STRING"）。schema_file_pathと相互に排他的です。
json_options.schema_file_path	JSON	String	—	.ddlスキーマファイルへのパス。schemaと相互に排他的です。Unity Catalog ボリュームパス (/Volumes/...) をサポートしています。
json_options.schema_evolution_mode	JSON	String	—	自動スキーマ推論のスキーマ進化モード。スキーマ進化モードを参照してください。
json_options.schema_hints	JSON	String	—	スキーマ推論に影響を与えるコンマ区切りの "column_name type" ペア (例: "id BIGINT, ts TIMESTAMP")。schema_evolution_mode が設定されている必要があります。スキーマヒントを使用したスキーマ推論のオーバーライドを参照してください。

オプション	適用対象	Type	デフォルト	説明
format	すべて	String	—	データのシリアル化形式。有効な値: STRING、JSON。STRINGには追加オプションは必要ありません。トランスフォーマーにjson_optionsが指定されていない場合、値はデフォルトでVariantとして解析されます。詳細については、Variant Data Formatを参照してください。
json_options.schema	JSON	String	—	Spark DDL形式のインラインスキーマ（例: "id BIGINT, name STRING"）。schema_file_pathと相互に排他的です。
json_options.schema_file_path	JSON	String	—	.ddlスキーマファイルへのパス。schemaと相互に排他的です。Unity Catalog ボリュームパス (/Volumes/...) をサポートしています。
json_options.schema_evolution_mode	JSON	String	—	自動スキーマ推論のスキーマ進化モード。スキーマ進化モードを参照してください。
json_options.schema_hints	JSON	String	—	スキーマ推論に影響を与えるコンマ区切りの "column_name type" ペア (例: "id BIGINT, ts TIMESTAMP")。schema_evolution_mode が設定されている必要があります。スキーマヒントを使用したスキーマ推論のオーバーライドを参照してください。

ファンアウト オプション

備考

プレビュー

この機能はプライベート プレビュー段階です。試用については、Databricksの連絡先にお問い合わせください。

ファンアウトオプションは、単一のKafkaソースからの各レコードを多数の宛先テーブルのいずれかにルーティングします。これらのオプションは、パイプライン定義内のスキーマオブジェクト (テーブルオブジェクトではない) の fanout_options で指定します。完全なパイプラインの例については、複数テーブルへのレコードのルーティング (ファンアウト) を、制約については、ファンアウトの制限 を参照してください。

オプション	Type	デフォルト	説明
fanout_by	String	—	必須。生のソースレコード（Kafka の key および value 列を含む）に対して評価され、その結果が宛先テーブル名を決定する SQL 式。 その値はテーブル名の最後のセグメントになります：{destination_catalog}.{destination_schema}.{value}。value 列はバイナリであるため、フィールドを抽出する前に、たとえば cast(value as string):event_type::string のように文字列にキャストします。式はNULL以外のSTRINGに解決され、その文字列は引用符やサニタイズなしでテーブル名セグメントとしてそのまま使用されるため、有効な引用符なしのテーブル識別子である必要があります。スペース、ドット、または引用符なしの識別子では無効なその他の文字を含む値は、書き込みに失敗します。完全に数字である値（たとえば、123）も同様です。値に文字またはアンダースコアも含まれている場合、先頭の数字は許可されます（例: 2024_events）。宛先テーブルは、まだ存在しない場合に自動的に作成されます。
transforms	Transformerのリスト	—	ルーティング後、宛先テーブルへの書き込み前に、ルーティングされた各レコードに適用される変換。ルーティング後に実行されるため、fanout_byが見る値には影響しません。最大1つの変換が許可されており、format: JSONを使用する必要があります。ファンアウト変換オプションを参照してください。

オプション	Type	デフォルト	説明
fanout_by	String	—	必須。生のソースレコード（Kafka の key および value 列を含む）に対して評価され、その結果が宛先テーブル名を決定する SQL 式。 その値はテーブル名の最後のセグメントになります：{destination_catalog}.{destination_schema}.{value}。value 列はバイナリであるため、フィールドを抽出する前に、たとえば cast(value as string):event_type::string のように文字列にキャストします。式はNULL以外のSTRINGに解決され、その文字列は引用符やサニタイズなしでテーブル名セグメントとしてそのまま使用されるため、有効な引用符なしのテーブル識別子である必要があります。スペース、ドット、または引用符なしの識別子では無効なその他の文字を含む値は、書き込みに失敗します。完全に数字である値（たとえば、123）も同様です。値に文字またはアンダースコアも含まれている場合、先頭の数字は許可されます（例: 2024_events）。宛先テーブルは、まだ存在しない場合に自動的に作成されます。
transforms	Transformerのリスト	—	ルーティング後、宛先テーブルへの書き込み前に、ルーティングされた各レコードに適用される変換。ルーティング後に実行されるため、fanout_byが見る値には影響しません。最大1つの変換が許可されており、format: JSONを使用する必要があります。ファンアウト変換オプションを参照してください。

ファンアウト変換オプション

transforms 内の各エントリでは、以下のオプションを使用します。ファンアウト変換では JSON 形式のみがサポートされています。

オプション	Type	デフォルト	説明
format	String	JSON	オプション。変換のシリアル化形式。ファンアウト変換でサポートされているのはJSONのみであり、省略された場合のdefaultです。STRING、AVRO、およびPROTOBUFはサポートされていません。
input_column	String	—	変換が読み取りおよび書き込みを行う列 (例: value)。このファンアウト JSON 変換は、この列をその場で解析します。

オプション	Type	デフォルト	説明
format	String	JSON	オプション。変換のシリアル化形式。ファンアウト変換でサポートされているのはJSONのみであり、省略された場合のdefaultです。STRING、AVRO、およびPROTOBUFはサポートされていません。
input_column	String	—	変換が読み取りおよび書き込みを行う列 (例: value)。このファンアウト JSON 変換は、この列をその場で解析します。

注記

output_column変換オプションは、ファンアウト変換には適用されません。ファンアウトJSON変換は、結果を常にその場でinput_columnに書き込みます。任意のoutput_columnの値は無視されます。

接続プロパティ

Catalog Explorer で Unity Catalog Kafka 接続を作成する際、認証方法に応じて次のプロパティを指定する必要があります。接続作成ステップについては、「Kafka接続の作成」を参照してください。

ユーザー名とパスワード

属性	説明
接続名	Unity Catalog内の接続の一意の名前。
接続タイプ	Select Kafka .
認証タイプ	ユーザー名とパスワード を選択します。
ユーザー名	Kafka クラスターでの認証に使用されるユーザー名。
パスワード	Kafka クラスターでの認証に使用されるパスワード。
ブートストラップサーバー	Kafkaクラスターのブートストラップサーバーアドレス (例: broker1:9092,broker2:9092)。
**スキーマレジストリURL**（オプション）	スキーマレジストリのURLです。
スキーマレジストリ API キー （オプション）	スキーマレジストリの API キーです。
スキーマレジストリ API シークレット（オプション）	スキーマレジストリの API シークレット。

属性	説明
接続名	Unity Catalog内の接続の一意の名前。
接続タイプ	Select Kafka .
認証タイプ	ユーザー名とパスワード を選択します。
ユーザー名	Kafka クラスターでの認証に使用されるユーザー名。
パスワード	Kafka クラスターでの認証に使用されるパスワード。
ブートストラップサーバー	Kafkaクラスターのブートストラップサーバーアドレス (例: broker1:9092,broker2:9092)。
**スキーマレジストリURL**（オプション）	スキーマレジストリのURLです。
スキーマレジストリ API キー （オプション）	スキーマレジストリの API キーです。
スキーマレジストリ API シークレット（オプション）	スキーマレジストリの API シークレット。

サービス資格情報

属性	説明
接続名	Unity Catalog内の接続の一意の名前。
接続タイプ	Select Kafka .
認証タイプ	「**サービス資格情報**」を選択します。
サービス認証情報	既存の Unity Catalog サービス認証情報を選択するか、 新しいサービス認証情報を作成 をクリックします。
ブートストラップサーバー	Kafkaクラスターのブートストラップサーバーアドレス (例: broker1:9092,broker2:9092)。
**スキーマレジストリURL**（オプション）	スキーマレジストリのURLです。
スキーマレジストリ API キー （オプション）	スキーマレジストリの API キーです。
スキーマレジストリ API シークレット（オプション）	スキーマレジストリの API シークレット。

属性	説明
接続名	Unity Catalog内の接続の一意の名前。
接続タイプ	Select Kafka .
認証タイプ	「**サービス資格情報**」を選択します。
サービス認証情報	既存の Unity Catalog サービス認証情報を選択するか、 新しいサービス認証情報を作成 をクリックします。
ブートストラップサーバー	Kafkaクラスターのブートストラップサーバーアドレス (例: broker1:9092,broker2:9092)。
**スキーマレジストリURL**（オプション）	スキーマレジストリのURLです。
スキーマレジストリ API キー （オプション）	スキーマレジストリの API キーです。
スキーマレジストリ API シークレット（オプション）	スキーマレジストリの API シークレット。

宛先テーブルスキーマ

Kafkaコネクタはストリーミングテーブル（追記専用）に書き込みます。宛先テーブルに書き込まれる列は、トランスフォーマーが構成されているかどうかによって異なります。

トランスフォーマーなし (生バイナリ)

key_transformerまたはvalue_transformerが設定されていない場合、宛先テーブルには以下の列が含まれます。

列	Type	説明
key	BINARY	Kafkaメッセージキーの生バイナリコンテンツ。
value	BINARY	Kafkaメッセージ値の生のバイナリコンテンツ。

列	Type	説明
key	BINARY	Kafkaメッセージキーの生バイナリコンテンツ。
value	BINARY	Kafkaメッセージ値の生のバイナリコンテンツ。

文字列トランスフォーマーを使用

トランスフォーマーでformat: STRINGが設定されている場合、対応する列はBINARYの代わりにSTRINGとして書き込まれます。

JSONトランスフォーマーを使用すると

トランスフォーマーで format: JSON が設定されている場合：

• json_optionsが指定されていない場合、列はVARIANTとして書き込まれます。
• json_options.schemaまたはjson_options.schema_file_pathが指定されている場合、JSONはスキーマと一致する型付き列に解析されます。
• json_options.schema_evolution_modeが設定されている場合、スキーマ推論が使用され、スキーマは自動的に進化します。

注記

Kafka メタデータ列（topic、partition、offset、timestamp、timestampType、headersを含む）は、ベータ版の宛先テーブルでは利用できません。