Kafka Connect は RDS バイナリログデータをどのように同期しますか?

1. 背景

私たちの事業開発では、次のようなシナリオによく遭遇します。

ビジネス更新データがデータベースに書き込まれる
ビジネス更新データは、下流の依存処理にリアルタイムで渡す必要がある

したがって、従来の処理アーキテクチャは次のようになります。

しかし、このアーキテクチャには多くの欠点もあります。プロジェクト内で大量のメッセージ送信コードを保守する必要があるのです。メッセージを追加または更新すると、多大なメンテナンスコストがかかります。したがって、これを処理するより良い方法は、次に示すように、データベースデータをストリーミングシステムに直接接続することです。

この記事では、E-MapReduce 上で RDS バイナリログを Kafka クラスターにリアルタイムで同期する方法を説明します。

2. 環境整備

実験ではVPCネットワーク環境を使用します。 VPC 環境ではデフォルトで以下のインスタンスが作成されます。

2.1 テストRDSデータベースを準備する

RDS インスタンスを作成し、バージョン 5.7 を選択します。ここでは、RDS の作成方法については詳しく説明しません。詳細な手順については、RDS のドキュメントを参照してください。作成後は、図のようになります。

2.2 Kafka クラスターの準備

E-MapReduce Kafka クラスターを作成し、バージョンとして EMR-3.11.0 を選択します。ここでは EMR-3.11.0 以降のバージョンを選択する必要があることに注意してください。そうしないと、Kafka Connect サービスはデフォルトでインストールおよび起動されません。詳細な作成プロセスについては、E-MapReduce のドキュメントを参照してください。作成後は、図のようになります。

注意: RDS インスタンスと E-MapReduce Kafka クラスターは同じ VPC 内に存在する必要があります。それ以外の場合は、2 つの VPC 間のネットワークを開く必要があります。

3. カフカコネクト

3.1 コネクタ

Kafka Connect は、Kafka と他のデータシステム間のデータ転送用のツールです。 Kafka ベースのデータパイプラインを実装し、上流と下流のデータソースを接続できます。必要なのは、データソースからのデータの読み取り/データソースへのデータの書き込み方法を具体的に実装するコネクタを Kafka Connect サービスで実行することだけです。 Confluent は多くのコネクタ実装を提供しており、ここからダウンロードできます。しかし、今日は Debezium が提供する MySQL Connector プラグイン (ダウンロードアドレス) を使用します。

このプラグインをダウンロードし、解凍したすべての jar パッケージを kafka lib ディレクトリにコピーします。注: これらの jar パッケージは、Kafka クラスター内のすべてのマシンにコピーする必要があります。

Kafka クラスターのサービスリストで Kafka Connect コンポーネントを再起動します。

3.2 コネクタを起動する

コネクタを作成する前に、いくつかの設定を行う必要があります。 Debezium MySQL Connector の主な構成項目の一部を次に示します。

Kafka クラスターにログインし、次のコマンドを使用してコネクタを設定および作成します。

この時点で、図に示すように、作成されたコネクタを確認できます。

3.3 注記

server_idとは何ですか? 「SELECT @@server_id;」を実行すると見つかります。 RDS で。

コネクタの作成時に接続エラーが発生する可能性があります。 RDS ホワイトリストに Kafka クラスターマシンへのアクセスが許可されていることを確認してください。

4 テスト

4.1 テーブルを作成する

しばらくすると、対応するトピックがKafkaクラスターに自動的に作成されます。