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オプションを参照してください。

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 が設定されている必要があります。スキーマヒントを使用したスキーマ推論のオーバーライドを参照してください。

接続プロパティ

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 .
認証タイプ	「**サービス資格情報**」を選択します。
サービス認証情報	既存の 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メッセージ値の生のバイナリコンテンツ。

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

トランスフォーマーで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を含む）は、ベータ版の宛先テーブルでは利用できません。