DataHub：Kafka資料移轉Datahub方案

前提

已經建立好Project和Topic，建立方式詳情見建立Topic樣本

操作步驟

1. 上傳kafka_mirror_datahub.tgz至源端Kafka伺服器並解壓

   tar -zxvf kafka_mirror_datahub.tgz

2. 進入config目錄，修改源端/目標端配置，設定檔說明如下:

   1. consumer.properties 源端kafka配置

   # 源端kafka server配置
   bootstrap.servers=xx:9092
   # consumer group 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=\"accessid\"\npassword=\"accesskey\";
   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。

   • sasl.jaas.config 用於指定 Kafka 用戶端或 Broker SASL 登入所需的登入模組、使用者名稱、密碼等敏感信 accessid、accesskey替換為使用者使用的ak資訊。

   • 更多配置項資訊請參照相容Kafka。

3. topic-map.properties topic 映射配置

   左邊為源端kafka topic名稱， 右邊為要寫入的DataHub topic，多組映射關係以行分割，Kafka的Topic映射之後為DataHub的project+topic, project和topic以 “.”分割 即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： 源端topic名稱，多個topic以|隔開 樣本 "a|b|c"。

   • --topic.mapping.file: topic 映射設定檔。

   • 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.ByteArraySerializer
      	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端資料是否寫入

   在資料匯流排頁面進入目標Project下的目標Topic（如kafkatest / test），在Shard列表頁簽確認Shard狀態為ACTIVE且存在最新資料時間。單擊右側抽樣，在抽樣面板中選擇Shard ID並設定採樣條數，單擊抽樣按鈕，即可在資料預覽表格中確認資料已成功寫入DataHub。