■ はじめに

　今日ってゆーかー昨日、
Kafka Connect の Connector 構成プロパティの指定がミスってて
2～3日ハマって迷惑かけたので、メモ。

【１】接続
　１）connection.url
　２）connection.user
　３）connection.password
　４）connection.attempts
　５）dialect.name
【２】書き込み
　１）insert.mode
　２）batch.size
　３）delete.enabled
【３】データマッピング
　１）table.name.format
　２）pk.mode
　３）pk.fields
　４）fields.whitelist
　５）db.timezone
【４】DDL サポート
　１）auto.create
　２）auto.evolve
　３）quote.sql.identifiers
【５】再試行
　１）max.retries
　２）retry.backoff.ms
【６】トラブル
　１）物凄く遅くDBに更新される現象が発生

【１】接続

https://docs.confluent.io/ja-jp/kafka-connectors/jdbc/10.0/sink-connector/sink_config_options.html#connection

１）connection.url

* JDBC URL

２）connection.user

* 接続先のユーザ名
 => 上のサイトの説明間違ってる？

３）connection.password

* 接続先のパスワード

４）connection.attempts

* 有効な JDBC 接続の取得を再試行する最大回数

５）dialect.name

* コネクターで使用する必要があるデータベース言語の名前

【２】書き込み

https://docs.confluent.io/ja-jp/kafka-connectors/jdbc/10.0/sink-connector/sink_config_options.html#writes

１）insert.mode

* Insertモード
* 指定可能な値: [insert、upsert、update]

[1] insert

* SQL の INSERT

[2] upsert

* 
* upsert モードを使用する場合は、コネクター構成に
　pk.mode プロパティと pk.fields プロパティを追加する必要がある
 => ★ここでハマった★

[3] update

* SQL の UPDATE

２）batch.size

* バッチ サイズ(メッセージグループサイズ)は、
　そのグループを送信する前になければならないバイト数
* デフォルト: 3000

補足：バッチ（batch）とは？

* Producerが、単一パーティションで送信されるメッセージグループのこと

https://learn.microsoft.com/ja-jp/azure/hdinsight/kafka/apache-kafka-performance-tuning
得られる効果は？

* batch.size を増やすと、
　ネットワークと IO 要求からのオーバーヘッドの処理が減るため、
　スループットを向上
 => 負荷が低く、バッチ サイズが大きくなると、
　　プロデューサーはバッチの準備が完了するのを待機するため、
　　Kafka の送信待ち時間が増える可能性がある
 => 負荷が高い場合は、スループットを向上させて待ち時間を減らすために、
　　バッチ サイズを増やすと、パフォーマンス向上を見込める

３）delete.enabled

* null レコード値を削除として扱うかどうか
* pk.mode を record_key に設定する必要がある
* デフォルト: false

【３】データマッピング

https://docs.confluent.io/ja-jp/kafka-connectors/jdbc/10.0/sink-connector/sink_config_options.html#data-mapping

１）table.name.format

* マップ先テーブル名のフォーマット制御文字列
* マップ元のトピック名を表すプレースホルダーとして「${topic}」を含めることができる
* 例：トピック「orders」の kafka_${topic} は、テーブル名「kafka_orders」にマップされる
* デフォルト: ${topic}

２）pk.mode

* プライマリキーモード
* 指定可能な値: [none、kafka、record_key、record_value]

[1] none

* キー指定しない

[2] kafka

* Apache Kafka® 座標???を PK として使用

[3] record_key

* レコードキーのフィールドを使用

[4] record_value

* レコード値のフィールドを使用

３）pk.fields

* プライマリキーのフィールド名のコンマ区切りのリスト
* pk.mode に依存

[1] none

* 何もしない

[2] kafka

* Kafka 座標を表す 3 つの値である必要がある
* 空の場合はデフォルトで __connect_topic,__connect_partition,__connect_offset になる

[3] record_key

* 指定されている場合は、目的のフィールドの抽出に使用
* 空の場合、キー構造体のすべてのフィールドが利用

[4] record_value

* 指定されている場合は、目的のフィールドの抽出に使用
* 空の場合は、値構造体からすべてのフィールドが利用

４）fields.whitelist

* レコード値フィールド名のコンマ区切りのリスト
* リストを設定した場合は、目的のフィールドのフィルター処理に使用
* 空の場合は、レコード値からすべてのフィールドが利用

５）db.timezone

* コネクターで時間ベースの値を挿入する場合に使用
* デフォルト: "UTC"

【４】DDL サポート

https://docs.confluent.io/ja-jp/kafka-connectors/jdbc/10.0/sink-connector/sink_config_options.html#ddl-support

１）auto.create

* 送信先テーブルが存在しない場合に、
　CREATE を自動的に作成するかどうかを指定
* デフォルト: false

２）auto.evolve

* レコードスキーマに関連する列がテーブルスキーマに存在しない場合に、
　ALTER を発行して自動的に追加するかどうかを指定
* デフォルト: false

３）quote.sql.identifiers

* SQL ステートメントで、テーブル名、列名、その他の識別子をいつクォートするかを指定

【５】再試行

https://docs.confluent.io/ja-jp/kafka-connectors/jdbc/10.0/sink-connector/sink_config_options.html#retries

１）max.retries

* エラー時に再試行する最大回数
* デフォルト: 10

２）retry.backoff.ms

* エラーの後、再試行するまでの待ち時間（ミリ秒）
* デフォルト: 3000

プログラムの超個人的なメモ

Memo for Programming.

【Kafka】Kafka Connect ～ Connector 構成プロパティ～

■ はじめに

目次

【１】接続

１）connection.url

２）connection.user

３）connection.password

４）connection.attempts

５）dialect.name

【２】書き込み

１）insert.mode

２）batch.size

３）delete.enabled

【３】データマッピング

１）table.name.format

２）pk.mode

３）pk.fields

４）fields.whitelist

５）db.timezone

【４】DDL サポート

１）auto.create

２）auto.evolve

３）quote.sql.identifiers

【５】再試行

１）max.retries

２）retry.backoff.ms

関連記事