ApsaraDB for ClickHouse：Menyinkronkan data dari instance ApsaraMQ for Kafka

Batasan

Hanya data dari instance ApsaraMQ for Kafka atau kluster Kafka yang dikelola sendiri yang dijalankan pada instance Elastic Compute Service (ECS) yang dapat disinkronkan ke instance ApsaraDB for ClickHouse.

Prasyarat

• Kluster tujuan ApsaraDB for ClickHouse:

  • Kluster tujuan telah dibuat. Instance sumber ApsaraMQ for Kafka dan kluster tujuan berada di wilayah yang sama serta menggunakan virtual private cloud (VPC) yang sama. Untuk informasi lebih lanjut, lihat Membuat Kluster ApsaraDB for ClickHouse.

  • Akun telah dibuat untuk mengakses database di kluster tujuan dan memiliki izin untuk melakukan operasi pada database. Untuk informasi lebih lanjut, lihat Manajemen Akun.

• Instance sumber ApsaraMQ for Kafka:

  • Topik telah dibuat. Untuk informasi lebih lanjut, lihat Langkah 1: Membuat Topik.

  • Grup konsumen telah dibuat. Untuk informasi lebih lanjut, lihat Langkah 2: Membuat Grup.

Catatan penggunaan

• Jika sebuah topik berlangganan oleh tabel eksternal Kafka ApsaraDB for ClickHouse, topik tersebut tidak dapat dikonsumsi oleh konsumen lainnya.

• Saat membuat tabel eksternal Kafka, tampilan materialisasi, dan tabel lokal, tipe bidang dalam ketiga tabel harus sesuai.

Prosedur

Contoh ini menunjukkan cara menyinkronkan data dari instance ApsaraMQ for Kafka ke tabel terdistribusi kafka_table_distributed di database default dari kluster ApsaraDB for ClickHouse yang menjalankan Edisi Kompatibel Komunitas.

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'host:port1,host:port2,host:port3',
    kafka_topic_list = 'topic_name1,topic_name2,...',
    kafka_group_name = 'group_name',
    kafka_format = 'data_format'[,]
    [kafka_row_delimiter = 'delimiter_symbol',]
    [kafka_num_consumers = N,]
    [kafka_thread_per_consumer = 1,]
    [kafka_max_block_size = 0,]
    [kafka_skip_broken_messages = N,]
    [kafka_commit_every_batch = 0,]
    [kafka_auto_offset_reset = N]

CREATE TABLE default.kafka_src_table ON CLUSTER `default`
(-- Bidang yang mendefinisikan skema tabel
    id Int32,
    name String               
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'alikafka-post-cn-****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-****1-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-****-3-vpc.alikafka.aliyuncs.com:9092',
    kafka_topic_list = 'testforCK',
    kafka_group_name = 'GroupForTestCK',
    kafka_format = 'CSV';

CREATE TABLE default.kafka_table_local ON CLUSTER default (
  id Int32,
  name String
) ENGINE = MergeTree()
ORDER BY (id);

CREATE MATERIALIZED VIEW <view_name> ON CLUSTER default TO <dest_table> AS SELECT * FROM <src_table>;

CREATE MATERIALIZED VIEW consumer ON CLUSTER default TO kafka_table_local AS SELECT * FROM kafka_src_table;

CREATE MATERIALIZED VIEW consumer ON CLUSTER default TO kafka_table_distributed AS SELECT * FROM kafka_src_table;

Langkah 6: Periksa apakah data telah disinkronkan

1. Kirim pesan ke topik instance ApsaraMQ for Kafka.

   1. Masuk ke Konsol ApsaraMQ for Kafka.

   2. Di halaman Instances, klik nama instance yang ingin Anda kelola.

   3. Di halaman Topics, temukan topik yang diperlukan dan pilih More > Send Message di kolom Actions.

   4. Di panel Send to Produce and Consume Message, tentukan isi pesan yang ingin Anda kirim.

      Dalam contoh ini, pesan 1,a dan 2,b dikirim.

   5. Klik OK.

2. Masuk ke database ApsaraDB for ClickHouse dan kueri tabel terdistribusi untuk memeriksa apakah data telah disinkronkan.

   Untuk informasi lebih lanjut tentang cara masuk ke database ApsaraDB for ClickHouse, lihat Hubungkan ke ClickHouse menggunakan DMS.

   Pernyataan contoh untuk memverifikasi data:

   Edisi Enterprise

   SELECT * FROM kafka_table_local; 

   Edisi Kompatibel Komunitas

   Pernyataan berikut menunjukkan cara mengkueri tabel terdistribusi.

   • Jika tabel tujuan adalah tabel lokal, Anda harus mengganti nama tabel terdistribusi dengan nama tabel lokal dalam pernyataan tersebut.

   • Jika Anda menggunakan kluster Edisi Kompatibel Komunitas dengan beberapa node, kami sarankan Anda mengkueri data dari tabel terdistribusi. Jika tidak, data hanya akan di-query dari satu node dalam kluster, yang menghasilkan hasil yang tidak lengkap.

   SELECT * FROM kafka_table_distributed; 

   Saat Anda menjalankan query dan menerima hasil yang sukses, ini menunjukkan bahwa data telah disinkronkan dari instance ApsaraMQ for Kafka ke kluster ApsaraDB for ClickHouse.

   Hasil contoh:

   ┌─id─┬─name─┐
   │  1 │  a   │
   │  2 │  b   │
   └────┴──────┘

   Jika hasil query berbeda dari yang Anda harapkan, Anda dapat melakukan pemecahan masalah lebih lanjut dengan mengikuti instruksi yang dijelaskan di (Opsional) Langkah 7: Lihat Status Konsumsi Tabel Eksternal Kafka.

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

FAQ

• Mengapa saya tidak bisa terhubung ke tabel eksternal seperti MySQL, HDFS, dan Kafka?

• Mengapa jumlah baris setelah impor dari Kafka tidak sesuai dengan ClickHouse?

• Mengapa data tidak bertambah setelah membuat tabel eksternal Kafka?