Realtime Compute for Apache Flink：Konektor ClickHouse

Informasi latar belakang

ClickHouse adalah sistem manajemen basis data berorientasi kolom yang digunakan untuk Pemrosesan Analitik Online (OLAP). Untuk informasi lebih lanjut, lihat Apa itu ClickHouse?.

Fitur

• Untuk tabel terdistribusi ClickHouse, data langsung ditulis ke tabel lokal ClickHouse yang sesuai dengan tabel terdistribusi tersebut.

• Untuk kluster ClickHouse yang diterapkan di Alibaba Cloud E-MapReduce (EMR), Anda dapat menggunakan semantik exactly-once.

Prasyarat

• Tabel ClickHouse telah dibuat. Untuk informasi lebih lanjut, lihat Buat Tabel Baru.

• Daftar putih dikonfigurasi untuk kluster ClickHouse.

  • Jika Anda menggunakan kluster ApsaraDB for ClickHouse dari Alibaba Cloud, konfigurasikan daftar putih sesuai petunjuk dalam Konfigurasikan Daftar Putih.

  • Jika Anda menggunakan kluster ClickHouse yang diterapkan di Alibaba Cloud EMR, konfigurasikan daftar putih sesuai petunjuk dalam Kelola Grup Keamanan.

  • Jika Anda menggunakan kluster ClickHouse mandiri yang di-hosting pada instance Elastic Compute Service (ECS), konfigurasikan daftar putih sesuai petunjuk dalam Ikhtisar.

  • Dalam kasus lain, konfigurasikan daftar putih mesin tempat kluster ClickHouse diterapkan untuk memastikan bahwa kluster ClickHouse dapat diakses oleh mesin tempat Realtime Compute for Apache Flink diterapkan.

  Catatan

  Untuk informasi lebih lanjut tentang cara melihat blok CIDR dari vSwitch tempat Realtime Compute for Apache Flink berada, lihat FAQ tentang manajemen dan operasi ruang kerja serta namespace.

Batasan

• Konektor ClickHouse tidak mendukung parameter sink.parallelism.

• Tabel hasil ClickHouse mendukung semantik at-least-once.

• Hanya Realtime Compute for Apache Flink yang menggunakan Ververica Runtime (VVR) 3.0.2 atau versi lebih baru yang mendukung Konektor ClickHouse.

• Hanya Realtime Compute for Apache Flink yang menggunakan VVR 3.0.3 atau VVR 4.0.7, atau versi minor lebih baru yang mendukung parameter ignoreDelete dalam klausa WITH.

• Hanya Realtime Compute for Apache Flink yang menggunakan VVR 4.0.10 atau versi lebih baru yang mendukung tipe data NESTED dari ClickHouse.

• Hanya Realtime Compute for Apache Flink yang menggunakan VVR 4.0.11 atau versi lebih baru yang memungkinkan Anda menulis data ke tabel lokal ClickHouse yang sesuai dengan tabel terdistribusi ClickHouse.

• Hanya Realtime Compute for Apache Flink yang menggunakan VVR 4.0.11 atau versi lebih baru yang menyediakan semantik exactly-once untuk menulis data ke tabel kluster ClickHouse yang diterapkan di Alibaba Cloud EMR. Semantik exactly-once tidak lagi dapat digunakan untuk menulis data ke tabel kluster ClickHouse EMR V3.45.1, atau versi minor lebih baru dari EMR V5.11.1 karena perubahan kemampuan EMR ClickHouse.

• Anda dapat mengatur parameter writeMode ke balance untuk menulis data secara merata ke tabel lokal ClickHouse hanya di Realtime Compute for Apache Flink yang menggunakan VVR 8.0.7 atau versi lebih baru.

• Hanya kluster ApsaraDB for ClickHouse Community-compatible Edition yang memungkinkan Anda menulis data ke tabel lokal ClickHouse.

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

Pemetaan tipe data

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

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

Contoh

• Contoh 1: Data ditulis ke tabel lokal ClickHouse pada node.

  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;

• Contoh 2: Data ditulis ke tabel terdistribusi ClickHouse.

  Tiga tabel lokal ClickHouse bernama local_table_test ada di node 192.XX.XX.1, 192.XX.XX.2, dan 192.XX.XX.3. Tabel terdistribusi ClickHouse bernama distributed_table_test dibuat berdasarkan tabel lokal ClickHouse.

  • Jika Anda ingin langsung menulis data dengan kunci yang sama ke tabel lokal ClickHouse yang sama pada node tertentu, jalankan pernyataan berikut:

    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;

  • Jika Anda tidak ingin secara manual menentukan node tempat data ditulis ke tabel lokal ClickHouse dalam parameter url, jalankan pernyataan berikut untuk memungkinkan sistem secara otomatis menyimpulkan node tabel lokal ClickHouse:

    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', -- Atur parameter url ke JDBC URL node tempat tabel terdistribusi ClickHouse berada. 
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'distributed_table_test', -- Atur parameter tableName ke nama tabel terdistribusi ClickHouse. 
      'shardWrite' = 'true',
      'inferLocalTable' = 'true', -- Atur parameter inferLocalTable ke true. 
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

FAQ

• Apakah saya bisa menarik kembali data yang diperbarui dari tabel hasil ApsaraDB for ClickHouse?

• Kapan saya bisa melihat data yang ditulis ke tabel sink ClickHouse di konsol ClickHouse?