ApsaraDB for SelectDB:Kafka kullanarak veri aktarma

Tekil mod

# Modify the broker address
bootstrap.servers=127.0.0.1:9092

name=test-selectdb-sink
connector.class=org.apache.doris.kafka.connector.DorisSinkConnector
topics=topic_test
doris.topic2table.map=topic_test:test_kafka_tbl
buffer.count.records=10000
buffer.flush.time=120
buffer.size.bytes=5000000
doris.urls=selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com
doris.http.port=8030
doris.query.port=9030
doris.user=admin
doris.password=****
doris.database=test_db
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

$KAFKA_HOME/bin/connect-standalone.sh -daemon $KAFKA_HOME/config/connect-standalone.properties $KAFKA_HOME/config/connect-selectdb-sink.properties

Dağıtık mod

# Modify the broker address
bootstrap.servers=127.0.0.1:9092
 
# Modify group.id. The ID must be the same for all workers in the same cluster.
group.id=connect-cluster

$KAFKA_HOME/bin/connect-distributed.sh -daemon $KAFKA_HOME/config/connect-distributed.properties

curl -i http://127.0.0.1:8083/connectors -H "Content-Type: application/json" -X POST -d '{
  "name":"test-selectdb-sink-cluster",
  "config":{
    "connector.class":"org.apache.doris.kafka.connector.DorisSinkConnector",
    "topics":"topic_test",
    "doris.topic2table.map": "topic_test:test_kafka_tbl",
    "buffer.count.records":"10000",
    "buffer.flush.time":"120",
    "buffer.size.bytes":"5000000",
    "doris.urls":"selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com",
    "doris.user":"admin",
    "doris.password":"***",
    "doris.database":"test_db",
    "doris.http.port":"8030",
    "doris.query.port":"9030",
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "value.converter":"org.apache.kafka.connect.json.JsonConverter"
  }
}'

Parametreler

Önkoşullar

1. Apache Kafka kümesi veya Confluent Cloud'u sürüm 2.4.0 veya üzeri olacak şekilde yükleyin. Bu örnekte tek düğümlü bir Kafka ortamı kullanılmaktadır.

   # Download and decompress the package
   wget https://archive.apache.org/dist/kafka/2.4.0/kafka_2.12-2.4.0.tgz
   tar -zxvf kafka_2.12-2.4.0.tgz
   cd kafka_2.12-2.4.0/
   bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 
   bin/kafka-server-start.sh -daemon config/server.properties

2. doris-kafka-connector-1.0.0.jar indirin ve JAR dosyasını KAFKA_HOME/libs dizinine yerleştirin.

3. Bir ApsaraDB for SelectDB örneği oluşturun. Daha fazla bilgi için bkz. Örnek oluşturma.

4. Bir ApsaraDB for SelectDB örneğine MySQL protokolünü kullanarak bağlanın. Daha fazla bilgi için bkz. Örneğe bağlanma.

5. Bir test veritabanı ve bir test tablosu oluşturun.

   1. Bir test veritabanı oluşturun.

      CREATE DATABASE test_db;

   2. Bir test tablosu oluşturun.

      USE test_db;
      CREATE TABLE employees (
          emp_no       int NOT NULL,
          birth_date   date,
          first_name   varchar(20),
          last_name    varchar(20),
          gender       char(2),
          hire_date    date
      )
      UNIQUE KEY(`emp_no`)
      DISTRIBUTED BY HASH(`emp_no`) BUCKETS 1;

Örnek 1: JSON verilerini senkronize etme

1. SelectDB sink'i yapılandırma

   Tekil modu örnek alarak Kafka config dizininde bir selectdb-sink.properties dosyası oluşturun ve aşağıdaki içeriği ekleyin:

   name=selectdb_sink
   connector.class=org.apache.doris.kafka.connector.DorisSinkConnector
   topics=test_topic
   doris.topic2table.map=test_topic:example_tbl
   buffer.count.records=10000
   buffer.flush.time=120
   buffer.size.bytes=5000000
   doris.urls=selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com
   doris.http.port=8030
   doris.query.port=9030
   doris.user=admin
   doris.password=***
   doris.database=test_db
   key.converter=org.apache.kafka.connect.storage.StringConverter
   value.converter=org.apache.kafka.connect.json.JsonConverter
   
   #Optional: Configure a dead-letter queue
   errors.tolerance=all
   errors.deadletterqueue.topic.name=test_error
   errors.deadletterqueue.context.headers.enable = true
   errors.deadletterqueue.topic.replication.factor=1

2. Kafka Connect'i başlatma

   bin/connect-standalone.sh -daemon config/connect-standalone.properties config/selectdb-sink.properties

Örnek 2: Debezium kullanarak MySQL verilerini ApsaraDB for SelectDB ile senkronize etme

Birçok iş senaryosunda operasyonel bir veritabanından gelen verilerin gerçek zamanlı olarak senkronize edilmesi gerekir. Bu durum, veritabanının değişiklik verisi yakalama (CDC) mekanizmasının kullanılmasını gerektirir.

Debezium, Kafka Connect tabanlı bir CDC aracıdır ve MySQL, PostgreSQL, SQL Server, Oracle ve MongoDB gibi çeşitli veritabanlarına bağlanabilir. Veri değişikliklerini sürekli olarak standart bir biçimde bir Kafka konusuna gönderir ve böylece downstream sink'ler tarafından gerçek zamanlı olarak tüketilebilir hale getirir. Bu örnekte MySQL kullanılmaktadır.

1. Debezium'u indirin.

   wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.9.8.Final/debezium-connector-mysql-1.9.8.Final-plugin.tar.gz

2. İndirilen dosyayı çıkarın.

   tar -zxvf debezium-connector-mysql-1.9.8.Final-plugin.tar.gz

3. Çıkarılan tüm JAR dosyalarını KAFKA_HOME/libs dizinine yerleştirin.

4. MySQL kaynağını yapılandırın.

   Kafka config dizininde bir mysql-source.properties dosyası oluşturun ve aşağıdaki içeriği ekleyin:

   name=mysql-source
   connector.class=io.debezium.connector.mysql.MySqlConnector
   database.hostname=rm-bp17372257wkz****.rwlb.rds.aliyuncs.com
   database.port=3306
   database.user=testuser
   database.password=****
   database.server.id=1
   # A unique identifier for this client in Kafka
   database.server.name=test123
   # The databases and tables to sync. By default, all databases and tables are synced.
   database.include.list=test
   table.include.list=test.test_table
   database.history.kafka.bootstrap.servers=localhost:9092
   # The Kafka topic used to store database schema changes
   database.history.kafka.topic=dbhistory
   transforms=unwrap
   # See https://debezium.io/documentation/reference/stable/transformations/event-flattening.html
   transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
   # Record delete events
   transforms.unwrap.delete.handling.mode=rewrite

   Yapılandırmadan sonra varsayılan Kafka konu adı biçimi SERVER_NAME.DATABASE_NAME.TABLE_NAME olacaktır.

   Not

   Debezium yapılandırmaları için Debezium connector for MySQL sayfasına bakın.

5. ApsaraDB for SelectDB sink'ini yapılandırın.

   Kafka config dizininde bir selectdb-sink.properties dosyası oluşturun ve aşağıdaki içeriği ekleyin:

   name=selectdb-sink
   connector.class=org.apache.doris.kafka.connector.DorisSinkConnector
   topics=test123.test.test_table
   doris.topic2table.map=test123.test.test_table:test_table
   buffer.count.records=10000
   buffer.flush.time=120
   buffer.size.bytes=5000000
   doris.urls=selectdb-cn-4xl3jv1****-public.selectdbfe.rds.aliyuncs.com
   doris.http.port=8030
   doris.query.port=9030
   doris.user=admin
   doris.password=****
   doris.database=test
   key.converter=org.apache.kafka.connect.json.JsonConverter
   value.converter=org.apache.kafka.connect.json.JsonConverter
   
   #Optional: Configure a dead-letter queue
   #errors.tolerance=all
   #errors.deadletterqueue.topic.name=test_error
   #errors.deadletterqueue.context.headers.enable = true
   #errors.deadletterqueue.topic.replication.factor=1

   Not

   Verileri ApsaraDB for SelectDB ile senkronize ederken önceden bir veritabanı ve bir tablo oluşturmanız gerekir.

6. Kafka Connect'i başlatın.

   bin/connect-standalone.sh -daemon config/connect-standalone.properties config/mysql-source.properties config/selectdb-sink.properties

   Not

   Başlatmadan sonra logs/connect.log dosyasını kontrol ederek hizmetin başarıyla başlatıldığını doğrulayabilirsiniz.

Konnektör işlemleri

# Check the connector status
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/status -X GET
# Delete the current connector
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster -X DELETE
# Pause the current connector
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/pause -X PUT
# Resume the current connector
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/resume -X PUT
# Restart tasks within the connector
curl -i http://127.0.0.1:8083/connectors/test-selectdb-sink-cluster/tasks/0/restart -X POST

Daha fazla bilgi için Connect REST Interface sayfasına bakın.

Ölü mektup kuyruğu

Varsayılan olarak bir dönüştürme hatası konnektörün başarısız olmasına neden olur. Ancak konnektörü bu tür hataları atlayarak tolere edecek şekilde yapılandırabilirsiniz. Ayrıca hata ayrıntılarını, başarısız işlemi ve sorunlu kaydı daha sonraki analizler için bir ölü mektup kuyruğuna yazabilirsiniz.

errors.tolerance=all
errors.deadletterqueue.topic.name=test_error_topic
errors.deadletterqueue.context.headers.enable=true
errors.deadletterqueue.topic.replication.factor=1

Daha fazla bilgi için Error Reporting in Connect sayfasına bakın.

SSL özellikli bir Kafka kümesine bağlanma

Kafka Connect üzerinden SSL özellikli bir Kafka kümesine erişmek için Kafka broker'ının genel anahtarını doğrulamak üzere bir sertifika dosyası (client.truststore.jks) sağlamanız gerekir. Aşağıdaki yapılandırmayı connect-distributed.properties dosyanıza ekleyebilirsiniz:

# Connect worker
security.protocol=SSL
ssl.truststore.location=/var/ssl/private/client.truststore.jks
ssl.truststore.password=test1234
 
# Embedded consumer for sink connectors
consumer.security.protocol=SSL
consumer.ssl.truststore.location=/var/ssl/private/client.truststore.jks
consumer.ssl.truststore.password=test1234

Kafka Connect'i SSL özellikli bir Kafka kümesine bağlanacak şekilde yapılandırma hakkında daha fazla bilgi için Configure Kafka Connect sayfasına bakın.