8.4.4. Debezium SQL Server コネクター設定プロパティーの説明

name

デフォルトなし

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

connector.class

デフォルトなし

コネクターの Java クラスの名前。SQL Server コネクターには、常に io.debezium.connector.sqlserver.Sql Server Connector の値を使用してください。

tasks.max

1

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

database.hostname

デフォルトなし

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

database.port

1433

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

database.user

デフォルトなし

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

database.password

デフォルトなし

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

database.dbname

デフォルトなし

変更内容をストリームする SQL Server データベースの名前 database.names と一緒に使ってはいけません。

database.names

デフォルトなし

変更をストリームする SQL Server データベースの名前をコンマで区切って指定します。現在は、1 つのデータベース名のみサポートしています。database.dbname と併用しないでください。

このオプションは実験的なもので、本番では使用しないでください。これを使用すると、コネクターの動作がデフォルトの設定と互換性がなく、アップグレードやダウングレードのパスがありません。

database.server.name

デフォルトなし

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

table.include.list

デフォルトなし

Debezium がキャプチャーするテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。table.include.list に含まれていないテーブルはキャプチャーから除外されます。各識別子の形式は schemaName.tableName です。デフォルトでは、コネクターは指定のスキーマのシステム以外のテーブルをすべてキャプチャーします。table.exclude.list と併用しないでください。

table.exclude.list

デフォルトなし

キャプチャーから除外するテーブルの完全修飾テーブル識別子と一致する正規表現のコンマ区切りリスト (任意)。Debezium は table.exclude.list に含まれていないテーブルをすべてキャプチャーします。各識別子の形式は schemaName.tableName です。table.include.list と併用しないでください。

column.include.list

空の文字列

変更イベントメッセージの値に含まれる必要がある列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。プライマリーキー列は、値に含まれていない場合でもイベントのキーに常に含まれることに注意してください。また、column.exclude.list プロパティーも設定しないでください。

column.exclude.list

空の文字列

変更イベントメッセージの値から除外される必要がある列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は schemaName.tableName.columnName です。プライマリーキー列は、値から除外される場合でもイベントのキーに常に含まれることに注意してください。また、column.include.list プロパティーも設定しないでください。

column.mask.hash.hashAlgorithm.with.salt.salt

該当なし

文字ベースの列の完全修飾名と一致する正規表現のコンマ区切りリスト (任意)。列の完全修飾名の形式は `<schemaName>.<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、および実際のデータセットによっては、結果として得られるデータセットが完全にマスクされないことがあります。

time.precision.mode

adaptive

時間、日付、およびタイムスタンプは、異なる精度の種類で表すことができます。adaptive (デフォルト) は、データベース列の型を基にして、ミリ秒、マイクロ秒、またはナノ秒の精度値のいずれかを使用して、データベースの値と全く同じように時間とタイムスタンプをキャプチャーします。connect は、Kafka Connect の Time、Date、および Timestamp の組み込み表現を使用して、常に時間とタイムスタンプ値を表し、データベース列の精度に関わらず、ミリ秒の精度を使用します。時間的価値 を参照します。

decimal.handling.mode

precise

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

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

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

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

include.schema.changes

true

コネクターがデータベーススキーマの変更を、データベースサーバー ID と同じ名前の Kafka トピックに公開するかどうかを指定するブール値。各スキーマの変更は、データベース名が含まれるキーと、スキーマ更新を記述する JSON 構造である値で記録されます。これは、コネクターがデータベース履歴を内部で記録する方法には依存しません。デフォルトは true です。

tombstones.on.delete

true

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

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

false - delete イベントのみが出力されます。

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

column.truncate.to._length_.chars

該当なし

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

column.mask.with._length_.chars

該当なし

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

column.propagate.source.type

該当なし

出力された変更メッセージの該当するフィールドスキーマに元の型および長さをパラメーターとして追加する必要がある列の完全修飾名と一致する、正規表現のコンマ区切りリスト (任意)。スキーマパラメーター (__debezium.source.column.type、__debezium.source.column.length、および __debezium.source.column.scale) は、それぞれ元の型名と長さ (可変幅型) を伝播するために使用されます。シンクデータベースの対応する列を適切にサイズ調整するのに便利です。列の完全修飾名の形式は schemaName.tableName.columnName です。

datatype.propagate.source.type+

該当なし

出力された変更メッセージフィールドスキーマに元の型および長さをパラメーターとして追加する必要がある列のデータベース固有のデータ型名と一致する、正規表現のコンマ区切りリスト (任意)。スキーマパラメーター (__debezium.source.column.type、__debezium.source.column.length、および __debezium.source.column.scale) は、それぞれ元の型名と長さ (可変幅型) を伝播するために使用されます。シンクデータベースの対応する列を適切にサイズ調整するのに便利です。完全修飾データ型名の形式は schemaName.tableName.typeName です。SQL Server 固有のデータ型のリストは SQL Server データ型 を参照してください。

message.key.columns

該当なし

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

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

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

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

テーブルのキーを複数の列名に基づいて設定するには、列名の間にコンマを挿入します。

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

<schemaName>.<tableName>

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

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

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

テーブル inventory.customer では、列 pk1 と pk2 がメッセージキーとして指定されています。どのスキーマのpurchaseorders テーブルでも、pk3 と pk4 のカラムがメッセージキーとして使用されます。

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

binary.handling.mode

bytes

バイナリー (binary、varbinary) 列を変更イベントで表す方法を指定します。bytes はバイナリーデータをバイト配列として表します (デフォルト)。base64はバイナリーデータを base64 でエンコードされた文字列として表します。hex はバイナリーデータを 16 進エンコード (base16) 文字列として表します。

snapshot.mode

Initial

キャプチャーされたテーブルの構造 (および必要に応じてデータ) の最初のスナップショットを作成するモード。スナップショットが完了すると、コネクターはデータベースのやり直し (redo) ログから変更イベントの読み取りを続行します。以下の値がサポートされます。

snapshot.include.collection.list

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

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

このプロパティーは増分スナップショットの動作には影響しません。

snapshot.isolation.mode

repeatable_read

使用されるトランザクション分離レベルと、キャプチャー用に指定されたテーブルをコネクターがロックする期間を制御するモード。以下の値がサポートされます。

snapshot、read_committed、read_uncommitted の各モードでは、最初のスナップショット中に他のトランザクションがテーブルの行を更新することができません。exclusive と repeatable_readモードでは、同時更新ができません。

モードの選択は、データの整合性にも影響します。exclusive と snapshot モードのみが完全な整合性を保証します。つまり、最初のスナップショットとログのストリーミングが履歴の線形を保持します。repeatable_read および read_committed モードの場合は、たとえば、追加されたレコードが初回のスナップショットで 1 回、ストリーミングフェーズで 1 回 の計 2 回表示される可能性があります。しかし、この整合性レベルはデータのミラーリングであれば問題ないはずです。read_uncommitted の場合、データの整合性の保証はありません (一部のデータは損失または破損する可能性があります)。

event.processing.failure.handling.mode

fail

fail は、例外 (問題のあるイベントのオフセットを示す) を伝播させ、コネクターを停止させます。
warn は、問題のあるイベントをスキップし、問題のあるイベントのオフセットをログに記録します。
skip は、問題のあるイベントをスキップします。

poll.interval.ms

1000

各反復処理の実行中に新しい変更イベントが表示されるまでコネクターが待機する時間 (ミリ秒単位) を指定する正の整数値。デフォルトは 1000 ミリ秒 (1 秒) です。

max.queue.size

8192

データベースログから読み取られた変更イベントが Kafka に書き込まれる前に配置される、ブロッキングキューの最大サイズを指定する正の整数値。このキューは、Kafka への書き込みが遅い場合や Kafka が利用できない場合などに、CDC テーブルリーダーにバックプレシャーを提供できます。キューに発生するイベントは、このコネクターによって定期的に記録されるオフセットには含まれません。デフォルトは 8192 で、常に max.batch.size プロパティーに指定された最大バッチサイズよりも大きくする必要があります。

max.batch.size

2048

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

heartbeat.interval.ms

0

ハートビートメッセージが送信される頻度を制御します。
このプロパティーには、コネクターがメッセージをハートビートトピックに送信する頻度を定義する間隔 (ミリ秒単位) が含まれます。このプロパティーは、コネクターがデータベースから変更イベントを受信しているかどうかを確認するために使用できます。また、長期に渡り変更されるのはキャプチャーされていないテーブルのレコードのみである場合は、ハートビートメッセージを利用する必要があります。このような場合、コネクターはデータベースからログの読み取りを続行しますが、変更メッセージを Kafka に出力しないため、オフセットの更新が Kafka にコミットされません。これにより、コネクターの再起動後に再送信される変更イベントが増える可能性があります。このプロパティーを 0 に設定して、ハートビートメッセージが全く送信されないようにします。
デフォルトでは無効にされています。

heartbeat.topics.prefix

__debezium-heartbeat

ハートビートメッセージが送信されるトピックの命名を制御します。
トピックは、<heartbeat.topics.prefix>.<server.name> パターンに従って名前が付けられます。

snapshot.delay.ms

デフォルトなし

コネクターの起動後、スナップショットを取得するまで待機する間隔 (ミリ秒単位)。
クラスター内で複数のコネクターを開始する際に スナップショットが中断されないようにするために使用でき、コネクターのリバランスが実行される可能性があります。

snapshot.fetch.size

2000

スナップショットの実行中に各テーブルから 1 度に読み取る必要がある行の最大数を指定します。コネクターは、このサイズの複数のバッチでテーブルの内容を読み取ります。デフォルトは 2000 です。

query.fetch.size

デフォルトなし

指定のクエリーのデータベースのラウンドトリップごとにフェッチされる行の数を指定します。デフォルトは、JDBC ドライバーのデフォルトのフェッチサイズです。

snapshot.lock.timeout.ms

10000

スナップショットの実行時に、テーブルロックを取得するまで待つ最大時間 (ミリ秒単位) を指定する整数値。この時間間隔でテーブルロックを取得できないと、スナップショットは失敗します (スナップショット も参照してください)。
0 に設定すると、コネクターがロックを取得できない場合、直ちに失敗します。値 -1 は、無限に待つことを意味します。

snapshot.select.statement.overrides

デフォルトなし

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

プロパティーには、<schemaName>.<tableName> の形式で完全修飾テーブル名のコンマ区切りリストが含まれます。たとえば、

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

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

snapshot.select.statement.overrides.<schemaName>.<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 のレコードのみが含まれます。

sanitize.field.names

コネクター設定が、Avro を使用するように key.converter または value.converter パラメーターを明示的に指定する場合は true です。それ以外の場合のデフォルトは false です。

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

provide.transaction.metadata

false

true に設定すると、Debezium はトランザクション境界でイベントを生成し、トランザクションメタデータでデータイベントエンベロープをエンリッチします。

詳細は、トランザクションメタデータ を参照してください。

retriable.restart.connector.wait.ms

10000 (10 秒)

再試行可能なエラーが発生した後にコネクターを再起動するまで待機する時間 (ミリ秒単位)。

skipped.operations

デフォルトなし

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

signal.data.collection

デフォルト値なし

シグナルをコネクターに送信するために使用されるデータコレクションの完全修飾名 。
コレクション名の指定には次のフォーマットを使用します。
<databaseName>.<schemaName>.<tableName>

シグナル機能はテクノロジープレビュー機能です。

incremental.snapshot.chunk.size

1024

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

max.iteration.transactions

0

データベースの複数のテーブルからの変更をストリーミングする際に、メモリーの使用量を削減するために使用する、反復ごとの最大トランザクション数を指定します。0 (デフォルト) に設定すると、コネクターは現在の最大 LSN を変更をフェッチする範囲として使用します。ゼロより大きい値に設定すると、コネクターはこの設定で指定された n 番目の LSN を変更をフェッチする範囲として使用します。

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.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 です。