ApsaraDB for ClickHouse:Sinkronisasi data dari Kafka

Batasan

Anda hanya dapat menyinkronkan data dari instans ApsaraMQ for Kafka dan kluster Kafka yang dikelola sendiri yang di-deploy pada Instance ECS.

Prasyarat

• ApsaraDB for ClickHouse:

  • Kluster tujuan telah dibuat di wilayah dan VPC yang sama dengan instans ApsaraMQ for Kafka. Untuk informasi selengkapnya, lihat Buat kluster.

  • Akun database dengan izin yang diperlukan telah dibuat untuk kluster tujuan. Untuk informasi selengkapnya, lihat Manajemen Akun.

• ApsaraMQ for Kafka:

  • Topik telah dibuat.

  • Kelompok konsumen telah dibuat.

Catatan penggunaan

• Topik yang berlangganan oleh tabel eksternal Kafka ApsaraDB for ClickHouse tidak boleh memiliki konsumen lain.

• Saat membuat tabel eksternal Kafka, Tampilan yang di-materialisasi, dan tabel lokal, tipe bidang dari ketiga tabel tersebut harus sesuai.

Prosedur

Contoh berikut menyinkronkan data dari ApsaraMQ for Kafka ke tabel terdistribusi kafka_table_distributed di database default kluster Edisi Kompatibel Komunitas ApsaraDB for ClickHouse.

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`
(
    -- Definisikan bidang 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: Verifikasi sinkronisasi

1. Kirim pesan ke topik dalam instans ApsaraMQ for Kafka.

   1. Login ke Konsol ApsaraMQ for Kafka.

   2. Pada halaman Instance list, klik nama instans tujuan.

   3. Pada halaman Topics, temukan topik tujuan lalu pilih More > Send Message (Demo) di kolom Actions.

   4. Pada halaman Send and Consume Message with Quick Experience, masukkan Message Content.

      Contoh ini mengirim pesan 1,a dan 2,b.

   5. Klik OK.

2. Login ke kluster ApsaraDB for ClickHouse, kueri tabel terdistribusi, dan periksa apakah data telah tersinkronisasi.

   Untuk informasi lebih lanjut tentang cara login ke kluster ApsaraDB for ClickHouse, lihat Menghubungkan ke kluster ApsaraDB for ClickHouse menggunakan DMS.

   Gunakan pernyataan berikut untuk mengkueri dan memverifikasi data:

   Edisi Enterprise

   SELECT * FROM kafka_table_local; 

   Edisi yang kompatibel dengan Community

   Kode berikut memberikan contoh cara mengkueri tabel terdistribusi.

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

   • Jika Anda menggunakan kluster Edisi yang kompatibel dengan Community bertipe multi-node, kami sangat menyarankan agar Anda mengkueri tabel terdistribusi. Jika Anda langsung mengkueri tabel lokal, hanya data dari satu node yang akan dikembalikan, sehingga menghasilkan set hasil yang tidak lengkap.

   SELECT * FROM kafka_table_distributed; 

   Jika kueri mengembalikan hasil, maka sinkronisasi data dari Kafka ke ApsaraDB for ClickHouse berhasil.

   Hasil kueri adalah sebagai berikut.

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

   Jika hasil kueri tidak sesuai harapan, lanjutkan ke Langkah 7 (Opsional): Periksa status konsumsi tabel eksternal Kafka untuk melakukan pemecahan masalah lebih lanjut.

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

FAQ

• Mengapa tabel eksternal untuk sumber seperti MySQL, HDFS, dan Kafka tidak dapat dijangkau?

• Mengapa jumlah baris di ClickHouse tidak sesuai dengan jumlah message yang diimpor dari Kafka?

• Mengapa tidak ada data baru yang ditulis ke tabel eksternal Kafka setelah tabel tersebut dibuat?