Realtime Compute for Apache Flink:Konektor ClickHouse

Informasi latar belakang

ClickHouse adalah sistem manajemen database berorientasi kolom untuk Pemrosesan Analitik Online (OLAP). Untuk informasi selengkapnya, lihat Apa Itu ClickHouse?.

Fitur

• Menulis data langsung ke tabel lokal dari tabel terdistribusi ClickHouse.

• Mencapai semantik tepat-sekali saat menulis ke ClickHouse pada Alibaba Cloud E-MapReduce (EMR).

Prasyarat

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

• Konfigurasikan daftar putih.

  • Jika Anda menggunakan ApsaraDB for ClickHouse, lihat Konfigurasikan daftar putih.

  • Jika Anda menggunakan ClickHouse pada Alibaba Cloud E-MapReduce (EMR), lihat Kelola security group.

  • Jika Anda menggunakan kluster ClickHouse yang dikelola sendiri pada Instance ECS, lihat Ikhtisar security group.

  • Untuk semua konfigurasi lainnya, konfigurasikan daftar putih pada mesin tempat ClickHouse dideploy agar mengizinkan akses dari penerapan Realtime Compute for Apache Flink.

  Catatan

  Untuk melihat segmen jaringan vSwitch untuk Realtime Compute for Apache Flink, lihat Bagaimana cara mengonfigurasi daftar putih?.

Batasan

• Parameter sink.parallelism tidak didukung.

• Secara default, sink ClickHouse menyediakan semantik at-least-once.

• Konektor ClickHouse hanya didukung di Ververica Runtime (VVR) 3.0.2 dan versi setelahnya.

• Opsi ignoreDelete hanya didukung di VVR 3.0.3, VVR 4.0.7, dan versi setelahnya.

• Tipe data Nested ClickHouse hanya didukung di VVR 4.0.10 dan versi setelahnya.

• Penulisan langsung ke tabel lokal dari tabel terdistribusi hanya didukung di VVR 4.0.11 dan versi setelahnya.

• Semantik tepat-sekali untuk menulis ke ClickHouse pada EMR hanya tersedia di VVR 4.0.11 dan versi setelahnya. Namun, karena perubahan kemampuan produk, semantik ini tidak tersedia untuk ClickHouse pada EMR v3.45.1 dan versi setelah v5.11.1.

• Mode penulisan balance, yang mendistribusikan data secara merata ke node tabel lokal, hanya tersedia di VVR 8.0.7 dan versi setelahnya.

• Penulisan ke tabel lokal ClickHouse hanya didukung untuk Edisi kompatibel Komunitas ApsaraDB for ClickHouse.

• Saat Anda mendaftarkan katalog ClickHouse, nama database default tidak boleh mengandung tanda hubung (-). Jika tidak, validasi URL JDBC akan gagal.

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'
);

DENGAN parameter

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: Menulis ke tabel single-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: Menulis ke tabel terdistribusi.

  Asumsikan tabel terdistribusi bernama distributed_table_test dibuat dari tiga tabel lokal bernama local_table_test, yang berlokasi di node 192.XX.XX.1, 192.XX.XX.2, dan 192.XX.XX.3.

  • Jika Anda ingin Flink menulis data langsung ke tabel lokal dan mem-partisi data berdasarkan kunci, gunakan DDL 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 ingin Flink secara otomatis menemukan node tabel lokal alih-alih menentukannya secara manual dalam URL, gunakan DDL 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/default', -- URL JDBC dari salah satu node dalam kluster.
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'distributed_table_test', -- Nama tabel terdistribusi.
      'shardWrite' = 'true',
      'inferLocalTable' = 'true', -- Atur inferLocalTable ke true.
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

FAQ

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

• Visibilitas data di tabel sink ClickHouse