第8章 アプリケーションの Debezium コネクターの設定

デフォルトの動作では、Debezium コネクターは、名前がデータベースの名前から形成されているトピックと、変更が行われたテーブルの名前が Debezium コネクターによって各変更イベントレコードを送信することです。つまり、トピックは 1 つの物理テーブルのレコードを受け取ります。トピックで複数の物理テーブルのレコードを受信する場合、Debezium コネクターがそのトピックにレコードを再ルーティングするように設定する必要があります。

複数の物理テーブルのイベントレコードを同じトピックにルーティングするには、Debezium コネクターの Kafka Connect 設定で ByLogicalTableRouter 変換を設定します。ByLogicalTableRouter SMT の設定では、以下を判断する正規表現を指定する必要があります。

たとえば、.properties ファイルの設定は以下のようになります。

transforms=Reroute
transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter
transforms.Reroute.topic.regex=(.*)customers_shard(.*)
transforms.Reroute.topic.replacement=$1customers_all_shards

Debezium Change イベントキーは、テーブルのプライマリーキーを構成するテーブル列を使用します。複数の物理テーブルのレコードを 1 つのトピックにルーティングするには、イベントキーはそれらのすべてのテーブル全体で一意である必要があります。ただし、各物理テーブルには、そのテーブル内でのみ一意のプライマリーキーを設定できます。たとえば、テーブルの行には、myserver.mydb.customers_shard1 テーブルの行と同じキーの値がある場合があり myserver.mydb.customers_shard2 ます。

変更イベントレコードが同じトピックにあるテーブル全体で各イベントキーが一意になるように、ByLogicalTableRouter 変換によってフィールドが変更されイベントキーに挿入されます。デフォルトでは、挿入されたフィールドの名前はです __dbz__physicalTableIdentifier。挿入されたフィールドの値は、デフォルトの宛先トピック名です。

必要に応じて、別のフィールドをキーに挿入するように ByLogicalTableRouter 変換を設定できます。これを実行するには、key.field.name オプションを指定し、既存のプライマリーキーフィールド名と競合しないフィールド名に設定します。以下に例を示します。

transforms=Reroute
transforms.Reroute.type=io.debezium.transforms.ByLogicalTableRouter
transforms.Reroute.topic.regex=(.*)customers_shard(.*)
transforms.Reroute.topic.replacement=$1customers_all_shards
transforms.Reroute.key.field.name=shard_id

この例では、shard_id フィールドをルーティングされたレコードのキー構造に追加します。

キーの新しいフィールドの値を調整するには、以下のいずれかのオプションを設定します。

以下に例を示します。

transforms.Reroute.key.field.regex=(.*)customers_shard(.*)
transforms.Reroute.key.field.replacement=$2

この設定では、宛先のデフォルトのトピック名が以下のようになります。

myserver.mydb.customers_shard1
myserver.mydb.customers_shard2
myserver.mydb.customers_shard3

変換では、2 番目のキャプチャーグループ（シャード番号）の値をキーの新しいフィールドの値として使用します。この例では、挿入された key フィールドの値は、1、2、またはです 3。

テーブルにグローバルに一意の鍵が含まれ、キー構造を変更する必要がない場合は、key.enforce.uniqueness プロパティーをに設定し falseます。

...
transforms.Reroute.key.enforce.uniqueness=false
...