DataHub:Kafka Mirror Maker を使用した DataHub へのデータ移行

前提条件

プロジェクトとトピックを作成済みであることを確認してください。 詳細については、「トピックの作成」をご参照ください。

操作手順

1. kafka_mirror_datahub.tgz パッケージを移行元の Kafka サーバーにアップロードし、展開します。

   tar -zxvf kafka_mirror_datahub.tgz

2. config ディレクトリに移動して、移行元と移行先の設定ファイルを変更します。 各ファイルの説明は次のとおりです：

   1. consumer.properties ：ソース Kafka クラスターの設定ファイル。

   # ソース Kafka クラスターのサーバー設定
   bootstrap.servers=xx:9092
   # コンシューマーグループ ID
   group.id=test-consumer-group
   auto.offset.reset=earliest
   session.timeout.ms=60000
   heartbeat.interval.ms=40000
   ssl.endpoint.identification.algorithm=
   key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
   value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

   b. producer.properties ：移行先の DataHub サービスの設定ファイル。

   bootstrap.servers=dh-cn-zhangjiakou-pre.aliyuncs.com:9092
   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required\nusername=\"AccessKey ID\"\npassword=\"AccessKey Secret\";
   security.protocol=SASL_SSL
   sasl.mechanism=PLAIN
   key.serializer=org.apache.kafka.common.serialization.StringSerializer
   value.serializer=org.apache.kafka.common.serialization.StringSerializer
   compression.type=lz4
   # べき等性を無効化
   enable.idempotence=false

   パラメーターの説明

   • 移行先の bootstrap.servers のドメイン名一覧については、「Kafka 互換性」をご参照ください。 この例のエンドポイント dh-cn-zhangjiakou-pre.aliyuncs.com:9092 は、China (Zhangjiakou) リージョンに対応しています。

   • sasl.jaas.config パラメーターは、SASL 認証に必要なログインモジュールと認証情報を指定します。 AccessKey ID と AccessKey Secret をご自身の AccessKey 情報に置き換えてください。

   • 設定項目の詳細については、「Kafka 互換性」をご参照ください。

3. トピックマッピングのために topic-map.properties ファイルを設定します。

   このファイルは、移行元の Kafka トピックを移行先の DataHub トピックにマッピングします。 各行はマッピングルールを表し、左側が移行元のトピック名、右側が移行先のトピック名です。 移行先の DataHub トピックは Project.Topic 形式で指定します。 各マッピングルールは改行して記述します。

   topicname=testproject.testtopic
   topicname1=testproject1.testtopic1

4. ロギングのために log4j.properties ファイルを設定します。

   1. log4j.properties ファイルを作成します。

   2. 設定には次のテンプレートを使用します：

      1. log4j.rootLogger=INFO, stdout, file
         log4j.appender.stdout=org.apache.log4j.ConsoleAppender
         log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
         log4j.appender.stdout.layout.ConversionPattern=[%d{yyyy-MM-dd HH:mm:ss}] [%p] %m (%c:%L)%n
         log4j.appender.file=org.apache.log4j.DailyRollingFileAppender
         log4j.appender.file.DatePattern='.'yyyy-MM-dd
         log4j.appender.file.File=/opt/logs/mm1.log
         log4j.appender.file.layout=org.apache.log4j.PatternLayout
         log4j.appender.file.layout.ConversionPattern=[%d{yyyy-MM-dd HH:mm:ss}] [%p] %m (%c:%L)%n
         log4j.logger.kafka=INFO
         log4j.logger.org.apache.kafka=INFO
         log4j.logger.kafka.tools.MirrorMaker=INFO
         log4j.logger.org.apache.zookeeper=WARN

5. 移行スクリプトを実行します。

   Kafka のインストールルートディレクトリから次のスクリプトを実行し、ログ出力を確認します：

   パラメーターの説明

   • --consumer.config ：ソース Kafka クラスターの設定ファイル。

   • --producer.config ：移行先の DataHub サービスの設定ファイル。

   • --whitelist ：移行元のトピック名。 複数のトピックを指定するには、topicA|topicB|topicC のように、縦棒 (|) で区切ります。

   • --topic.mapping.file ：トピックマッピングの設定ファイル。

   • KAFKA_LOG4J_OPTS ：ログ設定ファイルへのパス。

   nohup KAFKA_LOG4J_OPTS="log4j.properties"  bin/kafka-mirror-maker.sh   --consumer.config config/consumer.properties   --producer.config config/producer.properties   --whitelist "mirrortest"   --topic.mapping.file /opt/kafka_2.12-3.7.2/config/topic-map.properties > /dev/null 2>&1 &

6. ログでエラーがないか確認します。 次の出力は、起動が成功したことを示します：

   1. [2025-08-06 17:27:41] [INFO] Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$:31)
      [2025-08-06 17:27:41] [INFO] Starting mirror maker (kafka.tools.MirrorMaker$:62)
      [2025-08-06 17:27:41] [INFO] Loaded topic mappings: mirrortest -> test_suyang.mirror (kafka.tools.MirrorMaker$:62)
      [2025-08-06 17:27:41] [INFO] ProducerConfig values:
      	acks = -1
      	batch.size = 16384
      	bootstrap.servers = [xxx]
      	buffer.memory = 33554432
      	client.dns.lookup = use_all_dns_ips
      	client.id = producer-1
      	compression.type = lz4
      	connections.max.idle.ms = 540000
      	delivery.timeout.ms = 2147483647
      	enable.idempotence = false
      	interceptor.classes = []
      	internal.auto.downgrade.txn.commit = false
      	key.serializer = class org.apache.kafka.common.serialization.StringSerializer
      	linger.ms = 0
      	max.block.ms = 9223372036854775807
      	max.in.flight.requests.per.connection = 1
      	max.request.size = 1048576
      	metadata.max.age.ms = 300000
      	metadata.max.idle.ms = 300000

7. DataHub コンソールにログインして、データが DataHub に書き込まれたことを確認します。

   [データバス] ページで、移行先のプロジェクト内のトピック (例： kafkatest/test) に移動します。 [Shard リスト] タブで、シャードのステータスが ACTIVE であり、最新のデータタイムスタンプが表示されていることを確認します。 [操作] 列の [サンプル] をクリックします。 表示されるパネルで Shard ID を選択し、取得するレコード数を指定して、[サンプル] をクリックします。 データプレビューテーブルで、データが正常に DataHub に書き込まれたことを確認します。