Data migration through Kafka Connect

ストリーム処理において、E-MapReduce は Kafka と他のシステム間のデータ同期や、Kafka クラスター間のデータ移行が必要になることがよくあります。本セクションでは、E-MapReduce 上の Kafka Connect を使用して、Kafka クラスター間のデータ同期またはデータ移行を迅速に実現する方法について説明します。

前提条件

•クラウドアカウントの登録が完了していること。詳細については、「クラウドアカウントの登録」を参照してください。
•E-MapReduce サービスが有効化されていること。
•クラウドアカウントの権限付与が完了していること。詳細については、「ロールの権限付与」を参照してください。

背景情報

Kafka Connect は、Kafka と他のシステム間のストリームデータを高速に転送するためのスケーラブルで信頼性の高いツールです。たとえば、Kafka Connect を使用すると、データベースの binlog データを取得して Kafka クラスターに同期し、データベースデータの移行を実現できます。Kafka クラスターはストリーム処理システムに接続できるため、データベースを下流のストリーム処理システムに間接的に接続することもできます。同時に、Kafka Connect は REST API インターフェースも提供しており、Kafka Connect の作成と管理を容易に行えます。

Kafka Connect には、スタンドアロンモードと分散モードの 2 つの動作モードがあります。スタンドアロンモードでは、すべてのワーカーが 1 つのプロセス内で実行されます。分散モードはスタンドアロンモードよりもスケーラブルでフォルトトレラントであり、最も一般的に使用されるモードであるため、本番環境で推奨されています。

本記事では、E-MapReduce 上の Kafka Connect の REST API インターフェースを使用して、Kafka クラスター間のデータ移行を実行する方法について説明します。Kafka Connect は分散モードを使用します。

ステップ 1 Kafka クラスターの作成

EMR 上にソース Kafka クラスターと宛先 Kafka クラスターを作成します。Kafka Connect は Task ノードにインストールされているため、宛先 Kafka クラスターには Task ノードを作成する必要があります。クラスター作成後、Task ノード上の Kafka Connect サービスはデフォルトで起動され、ポート番号は 8083 になります。

ソース Kafka クラスターと宛先 Kafka クラスターを同じセキュリティグループ内に作成することを推奨します。ソース Kafka クラスターと宛先 Kafka クラスターが異なるセキュリティグループにある場合、デフォルトでは両者のネットワークは相互に通信できません。この場合、両者のセキュリティグループを設定して、ネットワークが相互に通信できるようにする必要があります。

Alibaba Cloud E-MapReduce コンソールにログインします。

ソース Kafka クラスターと宛先 Kafka クラスターを作成します。詳細については、「クラスターの作成」を参照してください。
注意 宛先 Kafka クラスターの作成時に Task インスタンスを起動する必要があります。つまり、Task ノードを作成する必要があります。

ステップ 2 移行対象データの Topic 準備

ソース Kafka クラスター上に connect という名前の Topic を作成します。

SSH モードでソース Kafka クラスターのヘッダーノード (本例では emr-header-1) にログインします。
root ユーザーとして次のコマンドを実行し、connect という名前の Topic を作成します。
kafka-topics.sh --create --zookeeper emr-header-1:2181 --replication-factor 2 --partitions 10 --topic connect

注意 上記の操作完了後、ログインウィンドウは閉じずに保持してください。後で引き続き使用します。

ステップ 3 Kafka Connect コネクタの作成

宛先 Kafka クラスターの Task ノード上で、curl コマンドを使用して JSON データ経由で Kafka Connect コネクタを作成します。

SSH モードで宛先 Kafka クラスターの Task ノード (本セクションでは emr-worker-3) にログインします。
(オプション) Kafka Connect の設定をカスタマイズします。
宛先 Kafka クラスターの Kafka サービス設定ページにアクセスし、connect-distributed.properties 内の offset.storage.topic、config.storage.topic、status.storage.topic の 3 つの設定項目をカスタマイズします。詳細については、「コンポーネントパラメーター設定」を参照してください。
Kafka Connect は Topic 内にオフセット、設定、タスクステータスを保存します。Topic 名は offset.storage.topic、config.storage.topic、status.storage.topic の 3 つの設定項目に対応します。Kafka Connect はデフォルトのパーティションとレプリケーション係数を使用してこれら 3 つの Topic を自動作成します。パーティションとレプリケーション係数の設定項目は /etc/ecm/kafka-conf/connect-distributed.properties ファイルに保存されています。

root ユーザーとして次のコマンドを実行し、Kafka Connect を作成します。
curl -X POST -H "Content-Type: application/json" --data '{"name": "connect-test", "config": { "connector.class": "EMRReplicatorSourceConnector", "key.converter" : "org.apache.kafka.connect.converters.ByteArrayConverter", "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter", "src.kafka.bootstrap.servers": "${src- kafka-ip}:9092", "src.zookeeper.connect": "${src-kafka-curator-ip}:2181", "dest.zookeeper.connect": "${dest-kafka-curator-ip} :2181", "topic.whitelist": "${source-topic}", "topic.rename.format": "${dest-topic}", "src.kafka.max.poll.records": "300 " } }' http://emr-worker-3:8083/connectors
JSON データ内で、name フィールドは作成される Kafka Connect の名前を表します。本例では connect-test です。config フィールドは実際の状況に応じて設定する必要があり、主要な変数の説明は以下の通りです。

注意 上記の操作完了後、ログインウィンドウは閉じずに保持してください。後で引き続き使用します。

ステップ 4 Kafka Connect と Task ノードのステータス確認

Kafka Connect と Task ノードの情報を確認し、両方のステータスが正常であることを確認します。

宛先 Kafka クラスターの Task ノード (本セクションでは emr-worker-3) のログインウィンドウに戻ります。
root ユーザーとして次のコマンドを実行し、すべての Kafka Connect を表示します。
curl emr-worker-3:8083/connectors
image.png

root ユーザーとして次のコマンドを実行し、本例で作成した Kafka Connect (本例では connect-test) のステータスを確認します。
curl emr-worker-3:8083/connectors/connect-test/status
image.png
Kafka Connect (本例では connect-test) のステータスが RUNNING であることを確認します。

root ユーザーとして次のコマンドを実行し、Task ノード情報を表示します。
curl emr-worker-3:8083/connectors/connect-test/tasks

返された情報にエラーメッセージがないことを確認します。

ステップ 5 移行対象データの生成

コマンドを使用して、移行対象データをソースクラスターの connect Topic に送信します。

ソース Kafka クラスターのヘッダーノード (本例では emr-header-1) のログインウィンドウに戻ります。
root ユーザーとして次のコマンドを実行し、connect Topic にデータを送信します。
kafka-producer-perf-test.sh --topic connect --num-records 100000 --throughput 5000 --record-size 1000 --producer-props bootstrap.servers=emr-header-1:9092
image.png

ステップ 6 データ移行結果の確認

移行対象データを生成後、Kafka Connect によりデータは宛先クラスターの対応するファイル (本例では connect.replica) に自動で移行されます。

宛先 Kafka クラスターの Task ノード (本セクションでは emr-worker-3) のログインウィンドウに戻ります。

root ユーザーとして次のコマンドを実行し、データ移行が成功したかどうかを確認します。
kafka-consumer-perf-test.sh --topic connect.replica --broker-list emr-header-1:9092 --messages 100000

上記の返された結果から、ソース Kafka クラスターから送信された 100,000 件のデータが宛先 Kafka クラスターに移行されたことがわかります。

まとめ

本記事では、Kafka Connect を使用して Kafka クラスター間でデータを移行する方法について説明とデモンストレーションを行いました。Kafka Connect のより詳細な使用方法については、Kafka 公式サイト情報および REST API を参照してください。