🥇Debezium の紹介 - Apache Kafka 用 CDC

仕事をしていると、新しい技術ソリューションやソフトウェア製品に出会うことがよくありますが、それらに関する情報はロシア語のインターネット上にはほとんどありません。この記事では、Debezium を使用して 2 つの人気のある DBMS (PostgreSQL と MongoDB) から Kafka クラスターに CDC イベントを送信するように構成する必要があったときの、私の最近の実践例を使って、そのようなギャップの 1 つを埋めていきたいと思います。作業の結果として現れるこのレビュー記事が他の人にとって役立つことを願っています。

Debezium と CDC とは一般的に何ですか?

デベジウム — CDC ソフトウェアカテゴリの代表 (データ変更のキャプチャ)、より正確には、Apache Kafka Connect フレームワークと互換性のあるさまざまな DBMS 用のコネクタのセットです。

それオープンソースプロジェクト、 Apache License v2.0 に基づいてライセンスされており、Red Hat のスポンサーとなっています。開発は 2016 年から継続されており、現在、MySQL、PostgreSQL、MongoDB、SQL Server の DBMS を正式にサポートしています。 Cassandra と Oracle 用のコネクタもありますが、現時点では「早期アクセス」ステータスにあり、新しいリリースは下位互換性を保証しません。

CDC を従来のアプローチ (アプリケーションが DBMS から直接データを読み取る場合) と比較すると、その主な利点には、低遅延、高い信頼性、可用性を備えた行レベルでのデータ変更ストリーミングの実装が含まれます。最後の 2 つのポイントは、Kafka クラスターを CDC イベントのリポジトリとして使用することで達成されます。

もう 1 つの利点は、イベントの保存に単一のモデルが使用されるため、エンドアプリケーションはさまざまな DBMS の操作の微妙な違いを気にする必要がないことです。

最後に、メッセージブローカーを使用すると、データの変更を監視するアプリケーションが水平方向にスケールアウトできるようになります。同時に、データは DBMS から直接取得されるのではなく、Kafka クラスターから取得されるため、データソースへの影響は最小限に抑えられます。

Debezium アーキテクチャについて

Debezium の使用は、次のような単純なスキームに帰着します。

DBMS (データソースとして) → Kafka のコネクタ Connect → Apache Kafka → コンシューマ

例として、プロジェクト Web サイトからの図を次に示します。

ただし、この方式はシンクコネクタしか使用できないように見えるので、あまり好きではありません。

実際には状況は異なります。データレイクがいっぱいになるのです。 (上の図の最後のリンク) Debezium の使用方法はこれだけではありません。 Apache Kafka に送信されたイベントは、アプリケーションでさまざまな状況に対処するために使用できます。例えば：

無関係なデータをキャッシュから削除する。
通知の送信。
検索インデックスの更新。
ある種の監査ログ。
...

Java アプリケーションがあり、Kafka クラスターを使用する必要がない/使用する可能性がない場合は、次の方法を使用する可能性もあります。埋め込みコネクタ。明らかな利点は、追加のインフラストラクチャ (コネクタと Kafka の形式) が不要になることです。ただし、このソリューションはバージョン 1.1 以降非推奨となり、使用は推奨されなくなりました (将来のリリースではサポートが削除される可能性があります)。

この記事では、開発者が推奨するフォールトトレランスとスケーラビリティを提供するアーキテクチャについて説明します。

コネクタ構成

最も重要な値であるデータの変化の追跡を開始するには、以下が必要です。

データソース。バージョン 5.7 以降の MySQL、PostgreSQL 9.6 以降、MongoDB 3.2 以降 (全リスト);
Apache Kafka クラスター。
Kafka Connect インスタンス (バージョン 1.x、2.x)。
構成された Debezium コネクタ。

最初の 2 つの点に取り組みます。 DBMS と Apache Kafka のインストールプロセスは、この記事の範囲外です。ただし、すべてをサンドボックスにデプロイしたい人のために、サンプルを含む公式リポジトリには既製のリポジトリがあります。 docker-compose.yaml.

最後の 2 つの点について詳しく説明します。

0.カフカコネクト

この記事のここおよび以降のすべての構成例は、Debezium 開発者によって配布された Docker イメージのコンテキストで説明されています。これには、必要なすべてのプラグインファイル (コネクタ) が含まれており、環境変数を使用して Kafka Connect の構成を提供します。

Confluent から Kafka Connect を使用する場合は、必要なコネクタのプラグインを、で指定されたディレクトリに個別に追加する必要があります。 plugin.path または環境変数経由で設定 CLASSPATH。 Kafka Connect ワーカーとコネクタの設定は、ワーカー起動コマンドに引数として渡される構成ファイルによって決定されます。詳細については、を参照してください。ドキュメンテーション.

コネクタバージョンで Debeizum をセットアップするプロセス全体は 2 段階で実行されます。それぞれを見てみましょう。

1. Kafka Connect フレームワークのセットアップ

データを Apache Kafka クラスターにストリーミングするには、Kafka Connect フレームワークで次のような特定のパラメーターを設定します。

クラスターに接続するためのパラメーター、
コネクタ自体の構成が直接保存されるトピックの名前、
コネクタが実行されているグループの名前 (分散モードが使用されている場合)。

プロジェクトの公式 Docker イメージは、環境変数を使用した構成をサポートしています。これを使用します。そこで、画像をダウンロードします。

docker pull debezium/connect

コネクタを実行するために必要な環境変数の最小セットは次のとおりです。

BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 — クラスターメンバーの完全なリストを取得するための Kafka クラスターサーバーの初期リスト。
OFFSET_STORAGE_TOPIC=connector-offsets — コネクタが現在配置されている位置を保存するためのトピック。
CONNECT_STATUS_STORAGE_TOPIC=connector-status - コネクタとそのタスクのステータスを保存するためのトピック。
CONFIG_STORAGE_TOPIC=connector-config - コネクタ構成データとそのタスクの保存に関するトピック。
GROUP_ID=1 — コネクタタスクを実行できるワーカーのグループの識別子。分散利用時に必要 (配布済み) モード。

次の変数を使用してコンテナを起動します。

docker run 
  -e BOOTSTRAP_SERVERS='kafka-1:9092,kafka-2:9092,kafka-3:9092' 
  -e GROUP_ID=1 
  -e CONFIG_STORAGE_TOPIC=my_connect_configs 
  -e OFFSET_STORAGE_TOPIC=my_connect_offsets 
  -e STATUS_STORAGE_TOPIC=my_connect_statuses  debezium/connect:1.2

アブロに関する注意事項

デフォルトでは、Debezium はデータを JSON 形式で書き込みます。これはサンドボックスや少量のデータには許容されますが、高負荷のデータベースでは問題になる可能性があります。 JSON コンバーターの代わりに、次を使用してメッセージをシリアル化することもできます。アブロバイナリ形式に変換することで、Apache Kafka の I/O サブシステムの負荷が軽減されます。

Avro を使用するには、別のスキーマレジストリ (図を保存するため)。コンバーターの変数は次のようになります。

name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: VALUE_CONVERTER   
value: io.confluent.connect.avro.AvroConverter

Avro の使用とそのレジストリの設定の詳細については、この記事の範囲を超えています。さらに、わかりやすくするために、JSON を使用します。

2. コネクタ自体の設定

これで、ソースからデータを読み取るコネクタ自体の構成に直接移動できるようになります。

PostgreSQL と MongoDB という 2 つの DBMS のコネクタの例を見てみましょう。私には経験があり、違いはあります (小さいとはいえ、場合によっては重大です!)。

構成は JSON 表記で記述され、POST リクエストを使用して Kafka Connect にアップロードされます。

2.1.PostgreSQL

PostgreSQL のコネクタ構成の例:

{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "127.0.0.1",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "definitelynotpassword",
    "database.dbname" : "dbname",
    "database.server.name": "pg-dev",
    "table.include.list": "public.(.*)",
    "heartbeat.interval.ms": "5000",
    "slot.name": "dbname_debezium",
    "publication.name": "dbname_publication",
    "transforms": "AddPrefix",
    "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.AddPrefix.regex": "pg-dev.public.(.*)",
    "transforms.AddPrefix.replacement": "data.cdc.dbname"
  }
}

このセットアップ後のコネクタの動作原理は非常に簡単です。

初めて起動すると、構成で指定されたデータベースに接続し、モードで起動します。 初期スナップショット、条件付きを使用して取得されたデータの初期セットを Kafka に送信します。 SELECT * FROM table_name.
初期化が完了すると、コネクタは PostgreSQL WAL ファイルから変更を読み取るモードになります。

使用したオプションについて:

name — 以下で説明する構成が使用されるコネクタの名前。将来的には、この名前は、Kafka Connect REST API を介してコネクタを操作する (つまり、ステータスの表示/再起動/構成の更新) ために使用されます。
connector.class — 構成されたコネクタによって使用される DBMS コネクタクラス。
plugin.name — WAL ファイルからのデータを論理的にデコードするためのプラグインの名前。から選択可能 wal2json, decoderbuffs и pgoutput。最初の 2 つは、DBMS に適切な拡張機能をインストールする必要があります。 pgoutput PostgreSQL バージョン 10 以降の場合、追加の操作は必要ありません。
database.* — データベースに接続するためのオプション。 database.server.name — Kafka クラスター内のトピック名の形成に使用される PostgreSQL インスタンス名。
table.include.list — 変更を追跡するテーブルのリスト。形式で指定される schema.table_name;と一緒に使用することはできません table.exclude.list;
heartbeat.interval.ms — コネクタが特別なトピックにハートビートメッセージを送信する間隔 (ミリ秒)。
heartbeat.action.query — 各ハートビートメッセージの送信時に実行されるリクエスト (このオプションはバージョン 1.1 で登場)。
slot.name — コネクタによって使用されるレプリケーションスロットの名前。
publication.name - 名前出版物コネクタが使用する PostgreSQL 内。存在しない場合、Debezium は作成を試みます。接続を確立するユーザーにこのアクションに対する十分な権限がない場合、コネクタはエラーで終了します。
transforms ターゲットトピックの名前を変更する方法を正確に決定します。
- transforms.AddPrefix.type 正規表現を使用することを示します。
- transforms.AddPrefix.regex — ターゲットトピックの名前を再定義するマスク。
- transforms.AddPrefix.replacement - 私たちが再定義しているものを直接的に。

ハートビートと変換の詳細

デフォルトでは、コネクタはコミットされたトランザクションごとにデータを Kafka に送信し、その LSN (ログシーケンス番号) がサービストピックに記録されます。 offset。しかし、コネクタがデータベース全体ではなく、そのテーブルの一部 (データの更新が頻繁に行われないテーブル) のみを読み取るように構成されている場合はどうなるでしょうか?

コネクタは WAL ファイルを読み取り、監視しているテーブルへのトランザクションのコミットを検出しません。
したがって、トピック内またはレプリケーションスロット内の現在の位置は更新されません。
その結果、WAL ファイルがディスク上に保持され、ディスク容量が不足する可能性があります。

ここでオプションが役に立ちます。 heartbeat.interval.ms и heartbeat.action.query。これらのオプションをペアで使用すると、ハートビートメッセージが送信されるたびに、別のテーブルのデータ変更リクエストを実行できるようになります。したがって、コネクタが現在 (レプリケーションスロット内に) 配置されている LSN は常に更新されます。これにより、DBMS は不要になった WAL ファイルを削除できます。オプションがどのように機能するかについて詳しくは、ドキュメンテーション.

より注目に値するもう 1 つのオプションは、 transforms。利便性と美しさの方が重要ですが...

デフォルトでは、Debezium は次の命名ポリシーを使用してトピックを作成します。 serverName.schemaName.tableName。これは必ずしも便利とは限りません。オプション transforms 正規表現を使用して、テーブルのリスト、つまりイベントを特定の名前のトピックにルーティングする必要があるテーブルのリストを定義できます。

私たちの構成ではありがとう transforms 次のことが起こります: 監視対象データベースからのすべての CDC イベントは、次の名前のトピックに送られます。 data.cdc.dbname。それ以外の場合 (これらの設定がない場合)、Debezium はデフォルトで次のようなテーブルごとにトピックを作成します。 pg-dev.public.<table_name>.

コネクタの制限

PostgreSQL のコネクタ構成の説明を締めくくるには、その操作の次の機能と制限について説明する価値があります。

PostgreSQL 用コネクタの機能は、論理デコードの概念に依存しています。したがって、彼は データベース構造を変更するリクエストを追跡しません (DDL) - したがって、このデータはトピックには含まれません。
レプリケーションスロットを使用しているため、コネクタの接続が可能のみ主要な DBMS インスタンスに接続します。
コネクタがデータベースに接続するユーザーが読み取り専用権限を持っている場合は、最初の起動前にレプリケーションスロットを手動で作成し、データベースに公開する必要があります。

構成の適用

それでは、構成をコネクタにロードしましょう。

curl -i -X POST -H "Accept:application/json" 
  -H  "Content-Type:application/json"  http://localhost:8083/connectors/ 
  -d @pg-con.json

ダウンロードが成功し、コネクタが開始されたことを確認します。

$ curl -i http://localhost:8083/connectors/pg-connector/status 
HTTP/1.1 200 OK
Date: Thu, 17 Sep 2020 20:19:40 GMT
Content-Type: application/json
Content-Length: 175
Server: Jetty(9.4.20.v20190813)

{"name":"pg-connector","connector":{"state":"RUNNING","worker_id":"172.24.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.24.0.5:8083"}],"type":"source"}

素晴らしいです。セットアップが完了し、すぐに使用できるようになりました。ここで、コンシューマのふりをして Kafka に接続してみましょう。その後、テーブル内のエントリを追加および変更します。

$ kafka/bin/kafka-console-consumer.sh 
  --bootstrap-server kafka:9092 
  --from-beginning 
  --property print.key=true 
  --topic data.cdc.dbname

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', 'foo@bar.com');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

このトピックでは、次のように表示されます。

変更を含む非常に長い JSON

{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"int32",
        "optional":false,
        "field":"id"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Key"
  },
  "payload":{
    "id":1005
  }
}{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"before"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"after"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"version"
          },
          {
            "type":"string",
            "optional":false,
            "field":"connector"
          },
          {
            "type":"string",
            "optional":false,
            "field":"name"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"ts_ms"
          },
          {
            "type":"string",
            "optional":true,
            "name":"io.debezium.data.Enum",
            "version":1,
            "parameters":{
              "allowed":"true,last,false"
            },
            "default":"false",
            "field":"snapshot"
          },
          {
            "type":"string",
            "optional":false,
            "field":"db"
          },
          {
            "type":"string",
            "optional":false,
            "field":"schema"
          },
          {
            "type":"string",
            "optional":false,
            "field":"table"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"txId"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"lsn"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"xmin"
          }
        ],
        "optional":false,
        "name":"io.debezium.connector.postgresql.Source",
        "field":"source"
      },
      {
        "type":"string",
        "optional":false,
        "field":"op"
      },
      {
        "type":"int64",
        "optional":true,
        "field":"ts_ms"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"id"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"total_order"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"data_collection_order"
          }
        ],
        "optional":true,
        "field":"transaction"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Envelope"
  },
  "payload":{
    "before":null,
    "after":{
      "id":1005,
      "first_name":"foo",
      "last_name":"bar",
      "email":"foo@bar.com"
    },
    "source":{
      "version":"1.2.3.Final",
      "connector":"postgresql",
      "name":"dbserver1",
      "ts_ms":1600374991648,
      "snapshot":"false",
      "db":"postgres",
      "schema":"public",
      "table":"customers",
      "txId":602,
      "lsn":34088472,
      "xmin":null
    },
    "op":"c",
    "ts_ms":1600374991762,
    "transaction":null
  }
}{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"int32",
        "optional":false,
        "field":"id"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Key"
  },
  "payload":{
    "id":1005
  }
}{
  "schema":{
    "type":"struct",
    "fields":[
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"before"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"int32",
            "optional":false,
            "field":"id"
          },
          {
            "type":"string",
            "optional":false,
            "field":"first_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"last_name"
          },
          {
            "type":"string",
            "optional":false,
            "field":"email"
          }
        ],
        "optional":true,
        "name":"data.cdc.dbname.Value",
        "field":"after"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"version"
          },
          {
            "type":"string",
            "optional":false,
            "field":"connector"
          },
          {
            "type":"string",
            "optional":false,
            "field":"name"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"ts_ms"
          },
          {
            "type":"string",
            "optional":true,
            "name":"io.debezium.data.Enum",
            "version":1,
            "parameters":{
              "allowed":"true,last,false"
            },
            "default":"false",
            "field":"snapshot"
          },
          {
            "type":"string",
            "optional":false,
            "field":"db"
          },
          {
            "type":"string",
            "optional":false,
            "field":"schema"
          },
          {
            "type":"string",
            "optional":false,
            "field":"table"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"txId"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"lsn"
          },
          {
            "type":"int64",
            "optional":true,
            "field":"xmin"
          }
        ],
        "optional":false,
        "name":"io.debezium.connector.postgresql.Source",
        "field":"source"
      },
      {
        "type":"string",
        "optional":false,
        "field":"op"
      },
      {
        "type":"int64",
        "optional":true,
        "field":"ts_ms"
      },
      {
        "type":"struct",
        "fields":[
          {
            "type":"string",
            "optional":false,
            "field":"id"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"total_order"
          },
          {
            "type":"int64",
            "optional":false,
            "field":"data_collection_order"
          }
        ],
        "optional":true,
        "field":"transaction"
      }
    ],
    "optional":false,
    "name":"data.cdc.dbname.Envelope"
  },
  "payload":{
    "before":{
      "id":1005,
      "first_name":"foo",
      "last_name":"bar",
      "email":"foo@bar.com"
    },
    "after":{
      "id":1005,
      "first_name":"egg",
      "last_name":"bar",
      "email":"foo@bar.com"
    },
    "source":{
      "version":"1.2.3.Final",
      "connector":"postgresql",
      "name":"dbserver1",
      "ts_ms":1600375609365,
      "snapshot":"false",
      "db":"postgres",
      "schema":"public",
      "table":"customers",
      "txId":603,
      "lsn":34089688,
      "xmin":null
    },
    "op":"u",
    "ts_ms":1600375609778,
    "transaction":null
  }
}

どちらの場合も、レコードは、変更されたレコードのキー (PK) と、変更の本質 (レコードが以前はどうであったか、その後どうなったか) で構成されます。

の場合 INSERT: 前の値 (before) に等しい null、その後 - 挿入された行。
の場合 UPDATEで： payload.before ラインの前の状態が表示されます。 payload.after — 変化の本質を備えた新しいもの。

2.2 モンゴDB

このコネクタは、標準の MongoDB レプリケーションメカニズムを使用し、プライマリ DBMS ノードの oplog から情報を読み取ります。

すでに説明した PgSQL 用のコネクタと同様に、ここでも最初の起動時にプライマリデータのスナップショットが取得され、その後コネクタは oplog 読み取りモードに切り替わります。

設定例：

{
  "name": "mp-k8s-mongo-connector",
  "config": {
        "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
        "tasks.max": "1",
        "mongodb.hosts": "MainRepSet/mongo:27017",
        "mongodb.name": "mongo",
        "mongodb.user": "debezium",
        "mongodb.password": "dbname",
        "database.whitelist": "db_1,db_2",
        "transforms": "AddPrefix",
        "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.AddPrefix.regex": "mongo.([a-zA-Z_0-9]*).([a-zA-Z_0-9]*)",
        "transforms.AddPrefix.replacement": "data.cdc.mongo_$1"
        }
  }

ご覧のとおり、前の例と比較して新しいオプションはありませんが、データベースとそのプレフィックスへの接続を担当するオプションの数だけが減りました。

設定 transforms 今回は次のことを行います: スキーマからターゲットトピックの名前を変換します。 <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

耐障害性

現代におけるフォールトトレランスと高可用性の問題は、これまで以上に深刻になっています。特にデータとトランザクションについて話している場合、この問題ではデータ変更の追跡は重要ではありません。原理的に何が問題になる可能性があるのか、そしてそれぞれのケースでDebeziumに何が起こるのかを見てみましょう。

オプトアウトオプションは 3 つあります。

Kafka 接続の失敗。 Connect が分散モードで動作するように構成されている場合、複数のワーカーが同じ group.id を設定する必要があります。その後、そのうちの 1 つが失敗した場合、コネクタは別のワーカーで再起動され、Kafka のトピック内の最後にコミットされた位置から読み取りを続けます。
Kafka クラスターとの接続が失われる。コネクタは、Kafka への送信に失敗した位置で読み取りを停止し、成功するまで定期的に再送信を試みます。
データソースが利用できない。コネクタは、構成に従ってソースへの再接続を試行します。デフォルトは 16 回の試行です。指数関数的バックオフ。 16 回目の試行が失敗すると、タスクは次のようにマークされます。 失敗した Kafka Connect REST インターフェイスを介して手動で再起動する必要があります。
- の場合 PostgreSQL データは失われません。レプリケーションスロットを使用すると、コネクタによって読み取られない WAL ファイルを削除できなくなります。この場合、コインにはマイナス面もあります。コネクタと DBMS 間のネットワーク接続が長時間中断されると、ディスク容量が不足し、DBMS の障害が発生する可能性があります。 DBMS全体。
- の場合 MySQL 接続が復元される前に、DBMS 自体によって binlog ファイルをローテーションできます。これにより、コネクタが障害状態になり、通常の動作を復元するには、初期スナップショットモードで再起動してバイナリログからの読み取りを続行する必要があります。
- オン MongoDBの。ドキュメントには、ログ/oplog ファイルが削除され、コネクタが中断した位置から読み取りを続行できない場合のコネクタの動作は、すべての DBMS で同じであると記載されています。コネクタが次の状態になることを意味します 失敗した モードで再起動する必要があります 初期スナップショット.
  ただし、例外もあります。コネクタが長時間切断され (または MongoDB インスタンスにアクセスできず)、その間に oplog がローテーションを行った場合、接続が復元されると、コネクタは最初に使用可能な位置から静かにデータを読み取り続けます。そのため、Kafka のデータの一部はノー当たります。

まとめ

Debezium は私にとって CDC システムの初めての経験であり、全体的に非常にポジティブです。このプロジェクトは、主要な DBMS のサポート、構成の容易さ、クラスタリングのサポート、および活発なコミュニティによって支持を集めました。実践に興味がある方は、ガイドを読むことをお勧めします。カフカコネクト и デベジウム.

Kafka Connect の JDBC コネクタと比較した場合、Debezium の主な利点は、変更が DBMS ログから読み取られるため、最小限の遅延でデータを受信できることです。 JDBC コネクタ (Kafka Connect から) は、監視対象テーブルに対して一定の間隔でクエリを実行し、(同じ理由で) データが削除されたときにメッセージを生成しません (存在しないデータをクエリするにはどうすればよいでしょうか?)。

同様の問題を解決するには、(Debezium に加えて) 次のソリューションに注意を払うことができます。

JDBC コネクタ Kafka 接続。
MySQL 専用のいくつかのソリューション:
Oracle GoldenGate, しかし、これは全く異なる「体重カテゴリー」です。

PS

私たちのブログもお読みください:

出所： habr.com