DataHub:Kafka Mirror Maker ile verileri DataHub'a geçirin

Ön koşullar

Bir Project ve Topic oluşturduğunuzu doğrulayın. Daha fazla bilgi için bkz. Topic oluşturma.

Yordam

1. kafka_mirror_datahub.tgz paketini kaynak Kafka sunucunuza yükleyin ve paketi açın.

   tar -zxvf kafka_mirror_datahub.tgz

2. Kaynak ve hedef yapılandırma dosyalarını değiştirmek için config dizinine gidin. Dosyalar aşağıda açıklanmıştır:

   1. a. consumer.properties: Kaynak Kafka kümesi için yapılandırma dosyası.

   # The server configuration for the source Kafka cluster
   bootstrap.servers=xx:9092
   # The 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: Hedef DataHub hizmeti için yapılandırma dosyası.

   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
   # Disable idempotence
   enable.idempotence=false

   Parametre açıklamaları

   • Hedef bootstrap.servers parametresi için alan adı listesini görmek üzere bkz. Kafka uyumluluğu. Bu örnekteki uç nokta olan dh-cn-zhangjiakou-pre.aliyuncs.com:9092 Çin (Zhangjiakou) bölgesine karşılık gelir.

   • sasl.jaas.config parametresi, SASL kimlik doğrulaması için gereken giriş modülünü ve kimlik bilgilerini belirtir. AccessKey ID ve AccessKey Secret değerlerini kendi AccessKey bilgilerinizle değiştirin.

   • Yapılandırma öğeleri hakkında daha fazla bilgi için bkz. Kafka uyumluluğu.

3. Konu eşlemesi için topic-map.properties dosyasını yapılandırın.

   Bu dosya, kaynak Kafka konularını hedef DataHub konularıyla eşler. Her satır, sol tarafın kaynak konu adını ve sağ tarafın hedefi temsil ettiği bir eşleme kuralını ifade eder. Hedef DataHub konusu Project.Topic biçiminde belirtilir. Her eşleme kuralını yeni bir satıra yerleştirin.

   topicname=testproject.testtopic
   topicname1=testproject1.testtopic1

4. Günlük kaydı için log4j.properties dosyasını yapılandırın.

   1. Bir log4j.properties dosyası oluşturun.

   2. Yapılandırmanız için aşağıdaki şablonu kullanın:

      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. Geçiş betiğini çalıştırın.

   Aşağıdaki betiği Kafka kurulumunuzun kök dizininden çalıştırın ve günlük çıktısını kontrol edin.

   Parametre açıklamaları

   • --consumer.config: Kaynak Kafka kümesi için yapılandırma dosyası.

   • --producer.config: Hedef DataHub hizmeti için yapılandırma dosyası.

   • --whitelist: Kaynak konu adları. Birden çok konu belirtmek için bunları dikey çubuk (|) ile ayırın; örneğin topicA|topicB|topicC.

   • --topic.mapping.file: Konu eşlemesi için yapılandırma dosyası.

   • KAFKA_LOG4J_OPTS: Günlük yapılandırma dosyasının yolu.

   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. Hataları kontrol etmek için günlüğü inceleyin. Aşağıdaki çıktı, başarılı bir başlangıcı gösterir:

   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. Verilerin DataHub'a yazıldığını doğrulamak için DataHub konsolunda oturum açın.

   Data Bus sayfasında, hedef Project içindeki hedef Topic'e (örneğin kafkatest/test) gidin. Shard List sekmesinde, parça durumunun ACTIVE olduğunu ve son veri zaman damgasının görüntülendiğini onaylayın. Actions sütunundaki Sample seçeneğine tıklayın. Görüntülenen panelde bir Shard ID seçin, alınacak kayıt sayısını belirtin ve Sample seçeneğine tıklayın. Veri önizleme tablosunda verilerin DataHub'a başarıyla yazıldığını doğrulayın.