5.5.5. Debezium MySQL コネクター設定プロパティーの説明

name

デフォルトなし

コネクターの一意名。同じ名前で再登録を試みると失敗します。このプロパティーはすべての Kafka Connect コネクターに必要です。

connector.class

デフォルトなし

コネクターの Java クラスの名前。MySQL コネクターに常に io.debezium.connector.mysql.MySqlConnector を指定します。

tasks.max

1

このコネクターのために作成する必要のあるタスクの最大数。MySQL コネクターは常に単一のタスクを使用するため、この値を使用しません。そのため、デフォルト値は常に許容されます。

database.hostname

デフォルトなし

MySQL データベースサーバーの IP アドレスまたはホスト名。

database.port

3306

MySQL データベースサーバーのポート番号 (整数)。

database.user

デフォルトなし

MySQL データベースサーバーへの接続時に使用する MySQL ユーザーの名前。

database.password

デフォルトなし

MySQL データベースサーバーへの接続時に使用するパスワード。

database.server.name

デフォルトなし

Debezium が変更をキャプチャーする特定の MySQL データベースサーバー/クラスターの namespace を識別および提供する論理名。論理名は、他のコネクター全体で一意となる必要があります。これは、このコネクターによって生成されるイベントを受信するすべての Kafka トピック名の接頭辞として使用されるためです。データベースサーバーの論理名には英数字とハイフン、ドット、アンダースコアのみを使用する必要があります。

+

database.server.id

random

このデータベースクライアントの数値 ID。MySQL クラスターで現在稼働しているすべてのデータベースプロセスで一意である必要があります。このコネクターは、MySQL データベースクラスターを (この一意の ID を持つ) 別のサーバーとして結合するため、binlog を読み取ることができます。デフォルトでは、5400 から 6400 までの乱数が生成されますが、値を明示的に設定することが推奨されます。

database.include.list

空の文字列

変更をキャプチャーするデータベースの名前と一致する正規表現のコンマ区切りリスト (任意)。コネクターは、名前が database.include.list にないデータベースの変更をキャプチャーしません。デフォルトでは、コネクターはすべてのデータベースの変更をキャプチャーします。また、database.exclude.list コネクター設定プロパティーは設定しないでください。

database.exclude.list

空の文字列

変更をキャプチャーしないデータベースの名前と一致する正規表現のコンマ区切りリスト (任意)。コネクターは、名前が database.exclude.list にないデータベースの変更をキャプチャーします。また、database.include.list コネクター設定プロパティーは設定しないでください。

table.include.list

空の文字列

変更をキャプチャーするテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。コネクターは table.include.list に含まれていないテーブルの変更をキャプチャーしません。各識別子の形式は databaseName.tableName です。デフォルトでは、コネクターは変更がキャプチャーされる各データベースのシステムでないすべてのテーブルの変更をキャプチャーします。また、table.exclude.list コネクター設定プロパティーは指定しないでください。

table.exclude.list

空の文字列

変更をキャプチャーしないテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。コネクターは table.exclude.list に含まれていないテーブルの変更をキャプチャーします。各識別子の形式は databaseName.tableName です。また、table.include.list コネクター設定プロパティーは指定しないでください。

column.exclude.list

空の文字列

変更イベントレコード値から除外する列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は databaseName.tableName.columnName です。

column.include.list

空の文字列

変更イベントレコード値に含める列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は databaseName.tableName.columnName です。

column.truncate.to.length.chars

該当なし

フィールド値が指定された文字数より長い場合に、変更イベントレコード値で値を省略する必要がある文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。単一の設定で、異なる長さの複数のプロパティーを設定できます。長さは正の整数である必要があります。列の完全修飾名の形式は databaseName.tableName.columnName です。

column.mask.with.length.chars

該当なし

変更イベントメッセージで、指定された数のアスタリスク (*) で設定されるフィールド値に値を置き換える必要のある文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。単一の設定で、異なる長さの複数のプロパティーを設定できます。それぞれの長さは正の整数またはゼロである必要があります。列の完全修飾名の形式は databaseName.tableName.columnName です。

column.mask.hash.hashAlgorithm.with.salt.salt; column.mask.hash.v2.hashAlgorithm.with.salt.salt

該当なし

文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は <databaseName>.<tableName>.<columnName> です。作成された変更イベントレコードでは、指定された列の値は仮名に置き換えられます。

仮名は、指定された hashAlgorithm と salt を適用すると得られるハッシュ化された値で設定されます。使用されるハッシュ関数に基づいて、参照整合性は維持され、列値は仮名に置き換えられます。サポートされるハッシュ関数は、Java Cryptography Architecture Standard Algorithm Name Documentation の MessageDigest section に説明されています。

以下の例では、CzQMA0cB5K が無作為に選択された salt になります。

column.mask.hash.SHA-256.with.salt.CzQMA0cB5K = inventory.orders.customerName, inventory.shipment.customerName

必要な場合は、仮名は自動的に列の長さに短縮されます。コネクター設定には、異なるハッシュアルゴリズムと salt を指定する複数のプロパティーを含めることができます。

使用される hashAlgorithm、選択された salt、および実際のデータセットによっては、結果として得られるデータセットが完全にマスクされないことがあります。

値が異なる場所やシステムでハッシュ化されている場合は、ハッシュ化ストラテジーバージョン 2 を使用する必要があります。

column.propagate.source.type

該当なし

出力された変更イベントレコードの該当するフィールドスキーマに元の型および長さをパラメーターとして追加する必要がある列の完全修飾名と一致する、正規表現のコンマ区切りリスト (任意)。以下のスキーマパラメーターは、それぞれ可変幅型の元の型名および長さを伝達するために使用されます。

__Debezium.source.column.type

__Debezium.source.column.length

__Debezium.source.column.scale

それぞれ元の型名と長さ (可変幅型の場合) を伝達するために使用されます。これは、シンクデータベースの対応する列を適切にサイズ調整するのに便利です。列の完全修飾名の形式は以下のいずれかになります。

databaseName.tableName.columnName

databaseName.schemaName.tableName.columnName

datatype.propagate.source.type

該当なし

出力された変更イベントレコードの該当するフィールドスキーマに元の型および長さをパラメーターとして追加する必要がある列のデータベース固有のデータ型名と一致する、正規表現のコンマ区切りリスト (任意)。以下のスキーマパラメーターは、それぞれ可変幅型の元の型名および長さを伝達するために使用されます。

__debezium.source.column.type

__debezium.source.column.length

__debezium.source.column.scale

それぞれ元の型名と長さ (可変幅型の場合) を伝達するために使用されます。これは、シンクデータベースの対応する列を適切にサイズ調整するのに便利です。完全修飾データ型名の形式は以下のいずれかになります。

databaseName.tableName.typeName

databaseName.schemaName.tableName.typeName

MySQL 固有のデータ型名のリストは、MySQL コネクターによるデータ型のマッピング方法 を参照してください。

time.precision.mode

adaptive_time_microseconds

時間、日付、およびタイムスタンプは、以下を含む異なる精度の種類で表すことができます。

adaptive_time_microseconds (デフォルト) は、データベース列の型を基にして、ミリ秒、マイクロ秒、またはナノ秒の精度値のいずれかを使用して、データベースの値と全く同じように日付、日時、およびタイムスタンプをキャプチャーします。


connect は、Kafka Connect の Time、Date、および Timestamp の組み込み表現を使用して、常に時間とタイムスタンプ値を表します。この組み込み表現は、データベース列の精度に関わらず、ミリ秒の精度を使用します。

decimal.handling.mode

precise

コネクターによる DECIMAL および NUMERIC 列の値の処理方法を指定します。

precise (デフォルト) はバイナリー形式で変更イベントに表される java.math.BigDecimal 値を使用して正確に表します。

double は double値を使用して表します。精度が失われる可能性はありますが、簡単に使用できます。

string は値をフォーマットされた文字列としてエンコードします。簡単に使用できますが、本来の型に関するセマンティック情報は失われます。

bigint.unsigned.handling.mode

long

変更イベントで BIGINT UNSIGNED 列を表す方法を指定します。可能な設定:

long は Java の long を使用して値を表します。これは、精度を提供しない可能性がありますが、コンシューマーでの使用が簡単です。通常、long が推奨設定となります。

precise は java.math.BigDecimal を使用して値を表します。値は、バイナリー表現と Kafka Connect の org.apache.kafka.connect.data.Decimal 型を使用して、変更イベントでエンコードされます。2^63 を超える値は long を使用して提供できないため、このような値を使用する場合はこの設定を使用します。

include.schema.changes

true

コネクターがデータベーススキーマの変更を、データベースサーバー ID と同じ名前の Kafka トピックに公開するかどうかを指定するブール値。各スキーマの変更はデータベース名が含まれるキーを使用して記録され、その値には DDL ステートメントが含まれます。これは、コネクターがデータベース履歴を内部で記録する方法には依存しません。

include.schema.comments

false

コネクターがメタデータオブジェクトでテーブルおよび列のコメントを解析して公開するかどうかを指定するブール値。このオプションを有効にすると、メモリー使用量に影響を及ぼします。論理スキーマオブジェクトの数およびサイズは、Debezium コネクターによって消費されるメモリーの量に大きく影響し、それぞれに大きな文字列データを追加すると、非常に高価になる可能性があります。

include.query

false

変更イベントを生成した元の SQL クエリーがコネクターに含まれる必要があるかどうかを指定するブール値。

このオプションを true に設定した場合は、MySQL の binlog_rows_query_log_events オプションを ON に設定する必要があります。include.query が true の場合、スナップショットプロセスによって生成されるイベントに対するクエリーは存在しません。

include.query を true に設定すると、変更イベントに元の SQL ステートメントを含めることで明示的に除外またはマスクされたテーブルまたはフィールドが公開される可能性があります。そのため、デフォルト設定は false です。

event.deserialization.failure.handling.mode

fail

binlog イベントのデシリアライズ中にコネクターがどのように例外に反応するかを指定します。

fail は例外を伝播し、問題のあるイベントとその binlog オフセットを示し、コネクターを停止させます。

warn は問題のあるイベントとその binlog オフセットをログに記録し、イベントをスキップします。

ignore は問題のあるイベントを渡し、何もログに記録しません。

inconsistent.schema.handling.mode

fail

内部スキーマ表現に存在しないテーブルに関連する binlog イベントに対してコネクターがどのように反応する必要があるかを指定します。つまり、内部表現はデータベースと一貫性がありません。

fail は例外を出力し、問題のあるイベントとその binlog オフセットを示し、コネクターを停止させます。

warn は問題のあるイベントとその binlog オフセットをログに記録し、イベントをスキップします。

skip は問題のあるイベントを渡し、何もログに記録しません。

max.batch.size

2048

このコネクターの反復処理中に処理される必要があるイベントの各バッチの最大サイズを指定する正の整数値。デフォルトは 2048 です。

max.queue.size

8192

ブロッキングキューが保持できるレコードの最大数を指定する正の整数値。Debezium はデータベースからストリームされたイベントを読み込む際、Kafka に書き込む前にブロッキングキューにイベントを配置します。ブロッキングキューは、コネクターが Kafka に書き込むよりも速くメッセージを取り込む場合、または Kafka が利用できなくなった場合に、データベースから変更イベントを読み込むためのバックプレッシャーを提供することができます。コネクターがオフセットを定期的に記録すると、キューに保持されるイベントは無視されます。max.queue.size の値を、max.batch.size の値よりも大きくなるように設定します。

max.queue.size.in.bytes

0

ブロッキングキューの最大容量をバイト単位で指定する長整数値。デフォルトでは、ブロックキューにはボリューム制限は指定されません。キューが使用できるバイト数を指定するには、このプロパティーを正の long 値に設定します。
max.queue.size も設定されている場合、キューのサイズがどちらかのプロパティーで指定された上限に達すると、キューへの書き込みがブロックされます。例えば、max.queue.size=1000、max.queue.size.in.bytes=5000 と設定した場合、キューに 1000 レコードが入った後、あるいはキュー内のレコードの量が 5000 バイトに達した後、キューへの書き込みがブロックされます。

poll.interval.ms

1000

コネクターがイベントのバッチの処理を開始する前に、新しい変更イベントの発生を待つ期間をミリ秒単位で指定する正の整数値。デフォルトは 1000 ミリ秒 (1 秒) です。

connect.timeout.ms

30000

コネクターが MySQL データベースサーバーへの接続を試行した後、タイムアウトするまでの最大の待機期間をミリ秒単位で指定する正の整数値。デフォルトは 30 秒です。

gtid.source.includes

デフォルトなし

MySQL サーバーで binlog の位置を見つけるために使用される GTID セットのソース UUID に一致する、正規表現のコンマ区切りリスト。これらの include パターンのいずれかに一致するソースを持つ GTID の範囲のみが使用されます。gtid.source.excludes の設定は指定しないでください。

gtid.source.excludes

デフォルトなし

MySQL サーバーで binlog の位置を見つけるために使用される GTID セットのソース UUID に一致する、正規表現のコンマ区切りリスト。これらすべての exclude パターンに一致しないソースを持つ GTID の範囲のみが使用されます。また、gtid.source.includes の値も指定しないでください。

tombstones.on.delete

true

削除 イベントの後に廃棄 (tombstone) イベントが続くかどうかを制御します。

true: 削除操作は、削除 イベントと後続の破棄 (tombstone) イベントで表されます。

false: 削除イベントのみ出力されます。

log compaction がトピックで有効になっている場合には、ソースレコードの削除後に廃棄 (tombstone) イベントを出力すると (デフォルト動作)、Kafka は削除された行のキーに関連するすべてのイベントを完全に削除できます。

message.key.columns

該当なし

指定のテーブルの Kafka トピックに公開する変更イベントレコードのカスタムメッセージキーを形成するためにコネクターが使用する列を指定する式のリスト。

デフォルトでは、Debezium はテーブルのプライマリーキー列を、出力するレコードのメッセージキーとして使用します。デフォルトの代わりに、またはプライマリーキーのないテーブルのキーを指定するには、1 つ以上の列をもとにカスタムメッセージキーを設定できます。

テーブルのカスタムメッセージキーを作成するには、テーブルとメッセージキーとして使用する列をリストします。各リストエントリーは以下の形式を取ります。

<fully-qualified_tableName>:<keyColumn>,<keyColumn>

複数の列名をベースにテーブルキーを作成するには、列名の間にコンマを挿入します。

各完全修飾テーブル名は、以下の形式の正規表現です。

<databaseName>.<tableName>

プロパティーには複数のテーブルのエントリーを含めることができます。セミコロンを使用して、リスト内のテーブルエントリーを区切ります。

以下の例では、テーブル inventory.customers と purchase.orders にメッセージキーを設定しています。

inventory.customers:pk1,pk2;(.*).purchaseorders:pk3,pk4

テーブル inventory.customer では、列 pk1 と pk2 がメッセージキーとして指定されています。データベースで purchaseorders テーブルは、pk3 および pk4 サーバーのコラムをメッセージキーとして使用します。

カスタムメッセージキーの作成に使用する列の数に制限はありません。ただし、一意の鍵を指定するために必要な最小数を使用することが推奨されます。

binary.handling.mode

bytes

バイナリー列 (例: blob、binary、varbinary) を変更イベントでどのように表すかを指定します。可能な設定:

bytes はバイナリーデータをバイト配列として表します。

base64 はバイナリーデータを base64 でエンコードされた文字列として表します。

hex は、バイナリーデータを 16 進数でエンコードされた (base16) 文字列として表します。

schema.name.adjustment.mode

avro

コネクターで使用されるメッセージコンバータとの互換性のために、スキーマ名をどのように調整するかを指定します。設定可能:

connect.keep.alive

true

MySQL サーバー/クラスターへの接続を確実に維持するために、別のスレッドを使用するかどうかを指定するブール値。

converters

デフォルトなし

コネクターが使用できる カスタムコンバーター インスタンスのシンボリック名のコンマ区切りリストを列挙します。
たとえば、boolean です。
このプロパティーは、コネクターがカスタムコンバーターを使用できるようにするために必要です。

コネクターに設定するコンバーターごとに、コンバーターインターフェイスを実装するクラスの完全修飾名を指定する .type プロパティーも追加する必要があります。.type プロパティーでは、以下の形式を使用します。

<converterSymbolicName>.type

以下に例を示します。

boolean.type: io.debezium.connector.mysql.converters.TinyIntOneToBooleanConverter

設定されたコンバータの動作をさらに制御したい場合は、1 つ以上の設定パラメーターを追加して、コンバータに値を渡すことができます。これらの追加設定パラメーターをコンバータに関連付けるには、パラメーター名の前にコンバーターのシンボル名を付けます。

例えば、boolean コンバーターが処理する列のサブセットを指定する selector パラメーターを定義するには、次のプロパティーを追加します。

boolean.selector=db1.table1.*, db1.table2.column1

table.ignore.builtin

true

組み込みシステムテーブルを無視するかどうかを指定するブール値。これは、テーブルの include および exclude リストに関係なく適用されます。デフォルトでは、システムテーブルは変更がキャプチャーされないように除外され、システムテーブルに変更が加えられてもイベントは生成されません。

database.ssl.mode

disabled

暗号化された接続を使用するかどうかを指定します。可能な設定:

disabled は暗号化されていない接続の使用を指定します。

preferred は、サーバーがセキュアな接続に対応している場合は暗号化された接続を確立します。サーバーがセキュアな接続に対応していない場合は、暗号化されていない接続にフォールバックします。

required は、暗号化された接続を確立し、何らかの理由で暗号化された接続を確立できない場合は失敗します。

verify_ca は required と同様に動作しますが、追加でサーバーの TLS 証明書を設定された認証局 (CA) 証明書に対して検証します。サーバー TLS 証明書が有効な CA 証明書と一致しない場合は失敗します。

verify_identity は verify_ca のように動作しますが、追加でサーバー証明書がリモート接続のホストと一致するかを検証します。

snapshot.mode

Initial

コネクターの起動時にスナップショットを実行するための基準を指定します。可能な設定は次のとおりです。

initial - コネクターは、論理サーバー名にオフセットが記録されていない場合にのみスナップショット を実行します。

initial_only - 論理サーバー名に対してオフセットが記録されてから停止した場合のみスナップショットを実行します。つまり、binlog から変更イベントを読み取りません。

when_needed - コネクターは、必要に応じて、コネクターは起動時にスナップショットを実行します。つまり、オフセットが使用できない場合や、以前に記録されたオフセットがサーバーが利用できない binlog の場所や GTID を指定する場合などです。

never - コネクターはスナップショットを使用しません。論理サーバー名での初回起動時に、コネクターは binlog の最初から読み取りします。この動作は注意して設定してください。これは、binlog にデータベースのすべての履歴が含まれることが保証されている場合のみ有効です。

schema_only - コネクターはデータではなく、スキーマのスナップショットを実行します。この設定は、トピックにデータの整合性スナップショットが含まれる必要がなく、コネクターの開始以降の変更のみが含まれる必要がある場合に便利です。

schema_only_recovery - これは、すでに変更をキャプチャーしているコネクターのリカバリー設定です。この設定により、コネクターを再起動すると、破損または損失したデータベース履歴トピックのリカバリーが可能になります。これを定期的に設定して、予想外に増加しているデータベース履歴トピックをクリーンアップすることができます。データベース履歴トピックは無期限に保持する必要があります。

snapshot.locking.mode

最小

コネクターがグローバル MySQL 読み込みロックを保持するかどうか、およびその期間を制御します。これにより、コネクターによるスナップショットの実行中にデータベースが更新されないようにします。可能な設定:

minimal - コネクターはスナップショットの最初の部分のみグローバル読み取りロックを保持します。その間、データベーススキーマとその他のメタデータを読み取ります。スナップショットの残りの作業では、各テーブルから全行を選択する必要があります。REPEATABLE READ トランザクションを使用すると、コネクターは一貫した方法でこれを行うことができます。これは、グローバル読み取りロックが保持されなくなり、その他の MySQL クライアントがデータベースを更新している場合でも該当します。

minimal_percona - コネクターは、スナップショットの最初の部分のみ グローバルバックアップロック を保持します。その間、コネクターはデータベーススキーマとその他のメタデータを読み取ります。スナップショットの残りの作業では、各テーブルから全行を選択する必要があります。REPEATABLE READ トランザクションを使用すると、コネクターは一貫した方法でこれを行うことができます。これは、グローバルバックアップロックが保持されなくなり、その他の MySQL クライアントがデータベースを更新している場合でも該当します。このモードはテーブルをディスクにフラッシュせず、長時間実行される読み取りによってブロックされず、Percona Server でのみ利用できます。

extended - スナップショットの実行中にすべての書き込みをブロックします。MySQL が REPEATABLE READ セマンティックから除外する操作を送信するクライアントがある場合は、この設定を使用します。

none - スナップショットの実行中にコネクターがテーブルロックを取得できないようにします。この設定はすべてのスナップショットモードで許可されますが、スナップショットの実行中にスキーマの変更がない場合に 限り、安全に使用できます。MyISAM エンジンで定義されたテーブルの場合、MyISAM によってテーブルロックが取得されるようにこのプロパティーが設定されていても、テーブルはロックされます。この動作は、行レベルのロックを取得する InnoDB エンジンの動作とは異なります。

snapshot.include.collection.list

table.include.listに指定したすべてのテーブル

スナップショットに含めるテーブルの完全修飾名 (<databaseName>.<tableName>) と一致する正規表現のコンマ区切りリスト (任意)。指定する項目は、コネクターの table.include.list プロパティーで名前を付ける必要があります。このプロパティーは、コネクターの snapshot.mode プロパティーが never 以外の値に設定されている場合にのみ有効になります。
このプロパティーは増分スナップショットの動作には影響しません。

snapshot.select.statement.overrides

デフォルトなし

スナップショットに追加するテーブル行を指定します。スナップショットにテーブルの行のサブセットのみを含める場合は、プロパティーを使用します。このプロパティーはスナップショットにのみ影響します。コネクターがログから読み取るイベントには影響しません。

<databaseName>.<tableName> の形式で完全修飾テーブル名のコンマ区切りリストを指定します。たとえば、

"snapshot.select.statement.overrides": "inventory.products,customers.orders"

をリスト内の各テーブルに対して、スナップショットを作成する場合には、その他の設定プロパティーを追加して、コネクターがテーブルで実行するように SELECT ステートメントを指定します。指定した SELECT ステートメントは、スナップショットに追加するテーブル行のサブセットを決定します。以下の形式を使用して、この SELECT ステートメントプロパティーの名前を指定します。

snapshot.select.statement.overrides.<databaseName>.<tableName>例: snapshot.select.statement.overrides.customers.orders.

例:

スナップショットにソフト削除以外のレコードのみを含める場合は、soft-delete 列 (delete_flag ) を含む customers.orders テーブルから、以下のプロパティーを追加します。

"snapshot.select.statement.overrides": "customer.orders",
"snapshot.select.statement.overrides.customer.orders": "SELECT * FROM [customers].[orders] WHERE delete_flag = 0 ORDER BY id DESC"

作成されるスナップショットでは、コネクターには delete_flag = 0 のレコードのみが含まれます。

min.row.count.to.stream.results

1000

スナップショットの実行中、コネクターは変更をキャプチャーするように設定されている各テーブルにクエリーを実行します。コネクターは各クエリーの結果を使用して、そのテーブルのすべての行のデータが含まれる読み取りイベントを生成します。このプロパティーは、MySQL コネクターがテーブルの結果をメモリーに格納するか、またはストリーミングを行うかを決定します。メモリーへの格納はすばやく処理できますが、大量のメモリーを必要とします。ストリーミングを行うと、処理は遅くなりますが、非常に大きなテーブルにも対応できます。このプロパティーの設定は、コネクターが結果のストリーミングを行う前にテーブルに含まれる必要がある行の最小数を指定します。

すべてのテーブルサイズチェックを省略し、スナップショットの実行中に常にすべての結果をストリーミングする場合は、このプロパティーを 0 に設定します。

heartbeat.interval.ms

0

コネクターがハートビートメッセージを Kafka トピックに送信する頻度を制御します。デフォルトの動作では、コネクターはハートビートメッセージを送信しません。

ハートビートメッセージは、コネクターがデータベースから変更イベントを受信しているかどうかを監視するのに便利です。ハートビートメッセージは、コネクターの再起動時に再送信する必要がある変更イベントの数を減らすのに役立つ可能性があります。ハートビートメッセージを送信するには、このプロパティーを、ハートビートメッセージの間隔をミリ秒単位で示す正の整数に設定します。

heartbeat.topics.prefix

__debezium-heartbeat

コネクターがハートビートメッセージを送信するトピックの名前を制御します。トピック名のパターンは次のようになります。

heartbeat.topics.prefix.server.name

たとえば、データベースサーバー名が fulfillment の場合、デフォルトのトピック名は __debezium-heartbeat.fulfillment になります。

heartbeat.action.query

デフォルトなし

コネクターがハートビートメッセージを送信するときにコネクターがソースデータベースで実行するクエリーを指定します。

たとえば、これを使用して、ソースデータベースで実行された GTID セットの状態を定期的にキャプチャできます。

INSERT INTO gtid_history_table(mysql.gtid_executed から * を選択)

database.initial.statements

デフォルトなし

トランザクションログを読み取る接続ではなく、データベースへの JDBC 接続が確立されたときに実行される SQL ステートメントのセミコロン区切りのリスト。SQL ステートメントでセミコロンを区切り文字としてではなく、文字として指定する場合は、2 つのセミコロン (;;) を使用します。

コネクターは独自の判断で JDBC 接続を確立する可能性があるため、このプロパティーはセッションパラメーターの設定専用です。DML ステートメントを実行するものではありません。

snapshot.delay.ms

デフォルトなし

コネクターの起動時にスナップショットを実行するまでコネクターが待つ必要がある間隔 (ミリ秒単位)。クラスターで複数のコネクターを起動する場合、このプロパティーは、コネクターのリバランスが行われる原因となるスナップショットの中断を防ぐのに役立ちます。

snapshot.fetch.size

デフォルトなし

スナップショットの実行中、コネクターは行のバッチでテーブルの内容を読み取ります。このプロパティーは、バッチの行の最大数を指定します。

snapshot.lock.timeout.ms

10000

スナップショットの実行時に、テーブルロックを取得するまで待つ最大時間 (ミリ秒単位) を指定する正の整数。コネクターがこの期間にテーブルロックを取得できないと、スナップショットは失敗します。Debezium MySQL コネクターによるデータベーススナップショットの実行方法 を参照してください。

enable.time.adjuster

true

コネクターによって 2 桁の西暦が 4 桁の西暦に変換されるかどうかを示すブール値。変換が完全にデータベースに委譲されている場合は、false に設定します。

MySQL では、2 桁または 4 桁の数値のいずれかで西暦の値を挿入できます。2 桁の値の場合は、値は 1970 - 2069 の範囲の年にマッピングされます。デフォルトの動作では、コネクターは変換を行いません。

sanitize.field.names

コネクターが key.converter または value.converter プロパティーを Avro コンバーターに設定する場合は true に設定します。
それ以外はfalse に設定します。

Avro の命名要件 に準拠するためにフィールド名がサニタイズされるかどうかを示します。

skipped.operations

デフォルトなし

ストリーミング中にスキップする操作タイプのコンマ区切りリスト。以下の値を使用できます (c は挿入/作成、u は更新、d は削除)。デフォルトでは、操作はスキップされません。

signal.data.collection

デフォルト値なし

シグナルをコネクターに送信するために使用されるデータコレクションの完全修飾名 。
以下の形式を使用してコレクション名を指定します。
<databaseName>.<tableName>

incremental.snapshot.allow.schema.changes

false

増分スナップショット時のスキーマの変更を許可します。有効にすると、コネクターは増分スナップショットの実行中にスキーマの変更を検出し、ロック DDL を回避するために現在のチャンクを再選択します。

プライマリーキーへの変更はサポートされず、増分スナップショットの実行時に実行された場合には誤った結果が生じる可能性があります。もう 1 つの制限は、スキーマの変更が列のデフォルト値のみに影響する場合、DDL が binlog ストリームから処理されるまで変更が検出されないことです。これはスナップショットイベントの値には影響しませんが、スナップショットイベントのスキーマにはデフォルト値が古くなっている可能性があります。

incremental.snapshot.chunk.size

1024

増分スナップショットのチャンクの実行中にコネクターがメモリーを取得して読み取る行の最大数。スナップショットは、サイズが大きいスナップショットの場合にはクエリーが少なくなるため、チャンクサイズを増やすと効率が上がります。ただし、チャンクサイズが大きい場合には、スナップショットデータのバッファーにより多くのメモリーが必要になります。チャンクサイズは、環境で最適なパフォーマンスを発揮できる値に、調整します。

provide.transaction.metadata

false

コネクターがトランザクション境界でイベントを生成し、トランザクションメタデータで変更イベントエンベロープを強化するかどうかを決定します。コネクターにこれを実行させる場合は true を指定します。詳細は、Transaction metadata を参照してください。

transaction.topic

${database.server.name}.transaction

コネクターがトランザクションのメタデータメッセージを送信するトピックの名前を制御します。プレースホルダー ${database.server.name} は、コネクターの論理名を参照するために使用できます。デフォルトは ${database.server.name}.transaction (例: dbserver1.transaction ) です。

database.history.kafka.topic

デフォルトなし

コネクターがデータベーススキーマの履歴を保存する Kafka トピックの完全名。

database.history.kafka.bootstrap.servers

デフォルトなし

Kafka クラスターへの最初の接続を確立するために コネクターが使用するホストとポートのペアの一覧。このコネクションは、コネクターによって以前に保存されたデータベーススキーマ履歴の取得や、ソースデータベースから読み取られる各 DDL ステートメントの書き込みに使用されます。各ペアは、Kafka Connect プロセスによって使用される同じ Kafka クラスターを示す必要があります。

database.history.kafka.recovery.poll.interval.ms

100

永続化されたデータのポーリングが行われている間にコネクターが起動/回復を待つ最大時間 (ミリ秒単位) を指定する整数値。デフォルトは 100 ミリ秒です。

database.history.kafka.query.timeout.ms

3000

Kafka 管理クライアントを使用してクラスター情報を取得する際に、コネクターが待機すべき最大ミリ秒数を指定する整数値です。

database.history.kafka.recovery.attempts

4

エラーでコネクターのリカバリーが失敗する前に、コネクターが永続化された履歴データの読み取りを試行する最大回数。データが受信されなかった場合に最大待機する時間は、recovery.attempts x recovery.poll.interval.ms です。

database.history.skip.unparseable.ddl

false

コネクターが不正または不明なデータベースのステートメントを無視するかどうか、または人が問題を修正するために処理を停止するかどうかを指定するブール値。安全なデフォルトは false です。スキップは、binlog の処理中にデータの損失や分割を引き起こす可能性があるため、必ず注意して使用する必要があります。

database.history.store.only.monitored.tables.ddl

今後のリリースで非推奨になり、削除される予定です。代わりに database.history.store.only.captured.tables.ddl を使用してください。

false

コネクターがすべての DDL ステートメントを記録するかどうかを指定するブール値

true は、変更が Debezium によってキャプチャーされるテーブルに関連する DDL ステートメントのみを記録します。変更がキャプチャーされるテーブルを変更すると、不足しているデータが必要になる可能性があるため、は、不足しているデータが必要になるため、注意して true に設定してください。

安全なデフォルトは false です。

database.history.store.only.captured.tables.ddl

false

コネクターがすべての DDL ステートメントを記録するかどうかを指定するブール値

true は、変更が Debezium によってキャプチャーされるテーブルに関連する DDL ステートメントのみを記録します。変更がキャプチャーされるテーブルを変更すると、不足しているデータが必要になる可能性があるため、は、不足しているデータが必要になるため、注意して true に設定してください。

安全なデフォルトは false です。

signal.kafka.topic

デフォルトなし

コネクターがアドホックシグナルについて監視する Kafka トピックの名前。

signal.kafka.bootstrap.servers

デフォルトなし

Kafka クラスターへの最初の接続を確立するために コネクターが使用するホストとポートのペアの一覧。各ペアは、Kafka Connect プロセスによって使用される同じ Kafka クラスターを示す必要があります。

signal.kafka.poll.timeout.ms

100

信号をポーリングするときにコネクターが待機する最大ミリ秒数を指定する整数値。デフォルトは 100 ミリ秒です。