E-MapReduce:MirrorMaker 2 (on Connect) を使用したクラスター間のデータ同期

背景

ユースケース

機能

タスクの実行方法

MM2 の詳細については、Apache Kafka ドキュメントをご参照ください。

前提条件

制限事項

ターゲットクラスターの Kafka バージョンは 2.12_2.4.1 以降である必要があります。

操作手順

1. ステップ 1：ターゲットクラスターでの Kafka Connect クラスターの作成
2. ステップ 2：MirrorMaker 2 コネクターの使用

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

1. E-MapReduce のタスクノードグループを作成します。
   E-MapReduce コンソールのemrdestターゲットクラスターの Nodes ページで、タスクノードグループを作成します。
   1. [Create Node Group] をクリックします。
   2. [Create Node Group] パネルで、次のパラメーターを設定します。他のパラメーターはデフォルト値を使用するか、必要に応じて設定します。

      パラメーター	説明
      [Node Group Type]	[TASK (Task Node Group)] を選択します。
      [Node Group Name]	名前を入力します。この例ではemr-taskを使用します。
      [Storage Configuration]	データディスクを選択します。

2. タスクノードグループをスケールアウトします。
   1. Nodes ページで、emr-taskノードグループを見つけ、[Actions] 列の スケールアウト をクリックします。
   2. 表示されるダイアログボックスで、追加するインスタンスの数を指定し、Terms of Service を選択します。
      この例では、1 つのインスタンスを追加します。必要に応じて、タスクインスタンスの数をスケールアウトできます。高可用性 Kafka Connect クラスターの場合、少なくとも 2 つのインスタンスを追加します。
   3. OK をクリックします。
3. KafkaConnect サービスのステータスを確認し、Kafka Connect クラスターが実行中であることを確かめます。
   1. ページ上部の Services をクリックします。
   2. Kafka サービスセクションで、Status をクリックします。
   3. Components セクションで、KafkaConnect のステータスが実行中であることを確認します。
      [Topology] リストで、KafkaConnect ノード task-1-1 (ロール：Task) のヘルスステータスが [Healthy] で、コンポーネントのステータスが [Running] であることを確認します。
4. SSH を使用してemrdestターゲットクラスターに接続します。詳細については、「クラスターへの接続」をご参照ください。
5. 次のコマンドを実行して、Kafka Connect REST サービスのステータスを確認します。

   curl -X GET http://task-1-1:8083| jq .

   出力は次のようになります。

     % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                    Dload  Upload   Total   Spent    Left  Speed
   100    91  100    91    0     0  13407      0 --:--:-- --:--:-- --:--:-- 15166
   {
     "version": "2.4.1",
     "commit": "42ce056344c5625a",
     "kafka_cluster_id": "6Z7IdHW4SVO1Pbql4c****"
   }

ステップ 2：MirrorMaker 2 コネクターの使用

1. MM2 コネクターの設定ファイルを準備します。
   次のファイルを準備します。

   • MirrorSourceConnector 設定ファイルを準備します。
     この例では、MirrorSourceConnector 設定ファイルの名前は mm2-source-connector.json です。次の例を使用して、ご使用の環境に合わせてパラメーター値を置き換えてください。設定オプションの詳細については、KIP-382 の関連セクションをご参照ください。

     {
       "name": "mm2-source-connector",
       "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
       "clusters": "emrsource,emrdest",
       "source.cluster.alias": "emrsource",
       "target.cluster.alias": "emrdest",
       "target.cluster.bootstrap.servers": "core-1-1:9092;core-1-2:9092;core-1-3:9092",
       "source.cluster.bootstrap.servers": "10.0.**.**:9092",
       "topics": "^foo.*",
       "tasks.max": "4",
       "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
       "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
       "replication.factor": "3",
       "offset-syncs.topic.replication.factor": "3",
       "sync.topic.acls.interval.seconds": "20",
       "sync.topic.configs.interval.seconds": "20",
       "refresh.topics.interval.seconds": "20",
       "refresh.groups.interval.seconds": "20",
       "consumer.group.id": "mm2-mirror-source-consumer-group",
       "producer.enable.idempotence":"true",
       "source.cluster.security.protocol": "PLAINTEXT",
       "target.cluster.security.protocol": "PLAINTEXT"
     }

     説明 この例のパラメーター：

     • source.cluster.bootstrap.servers：このパラメーターの値を、emrsource ソースクラスターの Kafka サービスのエンドポイントに置き換えてください。ソース Kafka クラスターと Kafka Connect クラスター間の通信を確保してください。
     • topics：このパラメーターは、ソースクラスターから foo で始まる名前のトピックをレプリケートします。
   • MirrorCheckpointConnector 設定ファイルを準備します。
     この例では、MirrorCheckpointConnector 設定ファイルの名前は mm2-checkpoint-connector.json です。次の例を使用して、ご使用の環境に合わせてパラメーター値を置き換えてください。設定オプションの詳細については、KIP-382 の関連セクションをご参照ください。

     {
         "name": "mm2-checkpoint-connector",
         "connector.class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
         "clusters": "emrsource,emrdest",
         "source.cluster.alias": "emrsource",
         "target.cluster.alias": "emrdest",
         "target.cluster.bootstrap.servers": "core-1-1:9092;core-1-2:9092;core-1-3:9092",
         "source.cluster.bootstrap.servers": "10.0.**.**:9092",
         "tasks.max": "1",
         "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
         "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
         "replication.factor": "3",
         "checkpoints.topic.replication.factor": "3",
         "emit.checkpoints.interval.seconds": "20",
         "source.cluster.security.protocol": "PLAINTEXT",
         "target.cluster.security.protocol": "PLAINTEXT"
       }

   • MirrorHeartbeatConnector 設定ファイルを準備します。
     この例では、MirrorHeartbeatConnector 設定ファイルの名前は mm2-heartbeat-connector.json です。次の例を使用して、ご使用の環境に合わせてパラメーター値を置き換えてください。設定オプションの詳細については、KIP-382 の関連セクションをご参照ください。

     {
         "name": "mm2-heartbeat-connector",
         "connector.class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
         "clusters": "emrsource,emrdest",
         "source.cluster.alias": "emrsource",
         "target.cluster.alias": "emrdest",
         "target.cluster.bootstrap.servers": "core-1-1:9092;core-1-2:9092;core-1-3:9092",
         "source.cluster.bootstrap.servers": "10.0.**.**:9092",
         "tasks.max": "1",
         "key.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
         "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
         "replication.factor": "3",
         "heartbeats.topic.replication.factor": "3",
         "emit.heartbeats.interval.seconds": "20",
         "source.cluster.security.protocol": "PLAINTEXT",
         "target.cluster.security.protocol": "PLAINTEXT"
       }

2. MirrorSourceConnector を使用します。
   1. Kafka Connect REST サービスで、mm2-source-connector.json ファイルを使用して MirrorSourceConnector タスクを作成します。

      curl -X PUT -H "Content-Type: application/json" --data @mm2-source-connector.json http://task-1-1:8083/connectors/mm2-source-connector/config

   2. 次のコマンドを実行して、mm2-source-connector のステータスを確認します。

      curl -s task-1-1:8083/connectors/mm2-source-connector/status | jq .

3. MirrorCheckpointConnector を使用します。
   1. Kafka Connect REST サービスで、mm2-checkpoint-connector.json ファイルを使用して MirrorCheckpointConnector タスクを作成します。

      curl -X PUT -H "Content-Type: application/json" --data @mm2-checkpoint-connector.json http://task-1-1:8083/connectors/mm2-checkpoint-connector/config

   2. 次のコマンドを実行して、mm2-checkpoint-connector のステータスを確認します。

      curl -s task-1-1:8083/connectors/mm2-checkpoint-connector/status | jq .

4. MirrorHeartbeatConnector を使用します。
   1. Kafka Connect REST サービスで、mm2-heartbeat-connector.json ファイルを使用して MirrorHeartbeatConnector タスクを作成します。

      curl -X PUT -H "Content-Type: application/json" --data @mm2-heartbeat-connector.json http://task-1-1:8083/connectors/mm2-heartbeat-connector/config

   2. 次のコマンドを実行して、mm2-heartbeat-connector のステータスを確認します。

      curl -s task-1-1:8083/connectors/mm2-heartbeat-connector/status | jq .

5. ターゲットクラスターで、次のコマンドを実行して MM2 に関連するトピックを表示します。

   kafka-topics.sh --list --bootstrap-server core-1-1:9092

   ターゲットクラスターには、次のトピックが作成されます。

   • emrsource.foo で始まるトピック：MirrorSourceConnector が作成します。

     これらは、ソースクラスター内の foo で始まるトピックに対応します。

   • emrsource.checkpoints.internal： MirrorCheckpointConnector が作成し、オフセットなどの情報を保存します。
   • heartbeats： MirrorHeartbeatConnector が作成します。