Realtime Compute for Apache Flink:Konektor ClickHouse

Prasyarat

Sebelum memulai, pastikan Anda telah:

• Tabel ClickHouse. Untuk informasi selengkapnya, lihat Buat tabel baru.

• Daftar putih yang dikonfigurasi untuk kluster ClickHouse Anda:

  • ApsaraDB for ClickHouse cluster: Configure the whitelist

  • Alibaba Cloud E-MapReduce (EMR) ClickHouse cluster: Manage security groups

  • ClickHouse yang dikelola sendiri pada Elastic Compute Service (ECS): Overview

  • Penerapan lainnya: Tambahkan Blok CIDR dari vSwitch yang digunakan oleh Realtime Compute for Apache Flink ke daftar izin mesin yang menjalankan kluster ClickHouse Anda. Untuk menemukan Blok CIDR tersebut, lihat FAQ about workspace and namespace management and operations.

Kemampuan konektor

Batasan

• Parameter sink.parallelism tidak didukung.

• Tabel hasil secara default mendukung semantik at-least-once.

• Memerlukan Ververica Runtime (VVR) versi 3.0.2 atau lebih baru.

• Parameter ignoreDelete memerlukan VVR 3.0.3, VVR 4.0.7, atau versi minor yang lebih baru dari salah satu versi tersebut.

• Tipe data NESTED memerlukan VVR 4.0.10 atau lebih baru.

• Menulis langsung ke tabel lokal ClickHouse yang berkorespondensi dengan tabel terdistribusi memerlukan VVR 4.0.11 atau lebih baru. Hanya kluster ApsaraDB for ClickHouse Community-compatible Edition yang mendukung fitur ini.

• Semantik exactly-once untuk kluster EMR ClickHouse memerlukan VVR 4.0.11 atau lebih baru. Fitur ini tidak lagi tersedia untuk kluster EMR V3.45.1 atau versi minor EMR yang lebih baru dari V5.11.1, karena adanya perubahan kemampuan pada EMR ClickHouse.

• Mode penulisan balance memerlukan VVR 8.0.7 atau lebih baru.

Sintaksis

CREATE TABLE clickhouse_sink (
  id INT,
  name VARCHAR,
  age BIGINT,
  rate FLOAT
) WITH (
  'connector' = 'clickhouse',
  'url' = '<yourUrl>',
  'userName' = '<yourUsername>',
  'password' = '<yourPassword>',
  'tableName' = '<yourTablename>',
  'maxRetryTimes' = '3',
  'batchSize' = '8000',
  'flushIntervalMs' = '1000',
  'ignoreDelete' = 'true',
  'shardWrite' = 'false',
  'writeMode' = 'partition',
  'shardingKey' = 'id'
);

Parameter dalam klausa WITH

Pilih mode penulisan

Parameter writeMode mengontrol cara catatan didistribusikan ke node tabel lokal saat menulis ke tabel terdistribusi ClickHouse dengan shardWrite=true.

writeMode=partition memerlukan ignoreDelete=true dan tidak kompatibel dengan exactlyOnce=true.

Pemetaan tipe data

CREATE TABLE visits (
  StartDate Date,
  Goals Nested
  (
    ID UInt32,
    OrderID String
  )
);

CREATE TABLE visits (
  StartDate DATE,
  `Goals.ID`      ARRAY<LONG>,
  `Goals.OrderID` ARRAY<STRING>
);

Contoh

CREATE TEMPORARY TABLE clickhouse_source (
  id   INT,
  name VARCHAR,
  age  BIGINT,
  rate FLOAT
) WITH (
  'connector'      = 'datagen',
  'rows-per-second' = '50'
);

CREATE TEMPORARY TABLE clickhouse_output (
  id   INT,
  name VARCHAR,
  age  BIGINT,
  rate FLOAT
) WITH (
  'connector' = 'clickhouse',
  'url'       = '<yourUrl>',
  'userName'  = '<yourUsername>',
  'password'  = '<yourPassword>',
  'tableName' = '<yourTablename>'
);

INSERT INTO clickhouse_output
SELECT id, name, age, rate
FROM clickhouse_source;

CREATE TEMPORARY TABLE clickhouse_source (
  id   INT,
  name VARCHAR,
  age  BIGINT,
  rate FLOAT
) WITH (
  'connector'       = 'datagen',
  'rows-per-second' = '50'
);

CREATE TEMPORARY TABLE clickhouse_output (
  id   INT,
  name VARCHAR,
  age  BIGINT,
  rate FLOAT
) WITH (
  'connector'  = 'clickhouse',
  'url'        = 'jdbc:clickhouse://192.XX.XX.1:3002,192.XX.XX.2:3002,192.XX.XX.3:3002/default',
  'userName'   = '<yourUsername>',
  'password'   = '<yourPassword>',
  'tableName'  = 'local_table_test',
  'shardWrite' = 'true',
  'writeMode'  = 'partition',
  'shardingKey' = 'name'
);

INSERT INTO clickhouse_output
SELECT id, name, age, rate
FROM clickhouse_source;

CREATE TEMPORARY TABLE clickhouse_source (
  id   INT,
  name VARCHAR,
  age  BIGINT,
  rate FLOAT
) WITH (
  'connector'       = 'datagen',
  'rows-per-second' = '50'
);

CREATE TEMPORARY TABLE clickhouse_output (
  id   INT,
  name VARCHAR,
  age  BIGINT,
  rate FLOAT
) WITH (
  'connector'      = 'clickhouse',
  'url'            = 'jdbc:clickhouse://192.XX.XX.1:3002/default', -- URL dari node mana pun yang menghosting tabel terdistribusi
  'userName'       = '<yourUsername>',
  'password'       = '<yourPassword>',
  'tableName'      = 'distributed_table_test',                    -- Nama tabel terdistribusi
  'shardWrite'     = 'true',
  'inferLocalTable' = 'true',                                     -- Temukan node tabel lokal secara otomatis
  'writeMode'      = 'partition',
  'shardingKey'    = 'name'
);

INSERT INTO clickhouse_output
SELECT id, name, age, rate
FROM clickhouse_source;

FAQ

Apakah saya dapat menarik kembali (menghapus) data yang diperbarui dari tabel hasil ApsaraDB for ClickHouse?

Atur ignoreDelete=false dan definisikan kunci primer dalam pernyataan DDL. Konektor kemudian akan mengeluarkan pernyataan ALTER DELETE saat menerima pesan retract. Perlu diperhatikan bahwa hal ini secara signifikan mengurangi throughput penulisan, dan tidak dapat digunakan bersamaan dengan writeMode=partition atau exactlyOnce=true. Untuk detail selengkapnya, lihat Can I retract the updated data from an ApsaraDB for ClickHouse result table?

Kapan data yang ditulis menjadi terlihat di ClickHouse?

ClickHouse menggunakan penggabungan part secara asinkron, sehingga data mungkin tidak langsung terlihat setelah penulisan selesai. Waktu tunda tergantung pada interval penggabungan yang dikonfigurasi di kluster ClickHouse Anda. Untuk detail selengkapnya, lihat When can I view the data that is written to a ClickHouse sink table in the ClickHouse console?