Realtime Compute for Apache Flink:ClickHouse connector

Arka plan bilgileri

ClickHouse, Çevrimiçi Analitik İşleme (OLAP) için kullanılan sütun tabanlı bir veritabanı yönetim sistemidir. Daha fazla bilgi için bkz. ClickHouse Nedir?.

Özellikler

• Verileri doğrudan ClickHouse dağıtılmış tablosunun yerel tablolarına yazın.

• Alibaba Cloud E-MapReduce (EMR) üzerindeki ClickHouse'a yazarken tam olarak bir kez (exactly-once) semantiği elde edin.

Ön koşullar

• Bir ClickHouse tablosu oluşturun. Daha fazla bilgi için bkz. Yeni tablo oluşturma.

• Bir izin listesini yapılandırın.

  • ApsaraDB for ClickHouse kullanıyorsanız bkz. İzin listesi yapılandırma.

  • Alibaba Cloud E-MapReduce (EMR) üzerinde ClickHouse kullanıyorsanız bkz. Güvenlik gruplarını yönetme.

  • Bir ECS örneğinde kendi yönettiğiniz bir ClickHouse kümesi kullanıyorsanız bkz. Güvenlik grubuna genel bakış.

  • Diğer tüm kurulumlar için, ClickHouse'un dağıtıldığı makinede Realtime Compute for Apache Flink dağıtımından erişime izin vermek üzere izin listesini yapılandırın.

  Not

  Realtime Compute for Apache Flink için vSwitch ağ segmentini görüntülemek üzere bkz. İzin listesi nasıl yapılandırılır?.

Sınırlamalar

• sink.parallelism parametresi desteklenmez.

• Varsayılan olarak ClickHouse sink en az bir kez (at-least-once) semantiği sağlar.

• ClickHouse connector yalnızca Ververica Runtime (VVR) 3.0.2 ve sonraki sürümlerde desteklenir.

• ignoreDelete seçeneği yalnızca VVR 3.0.3, VVR 4.0.7 ve sonraki sürümlerde desteklenir.

• ClickHouse Nested veri türü yalnızca VVR 4.0.10 ve sonraki sürümlerde desteklenir.

• Dağıtılmış bir tablonun yerel tablolarına doğrudan yazma işlemi yalnızca VVR 4.0.11 ve sonraki sürümlerde desteklenir.

• EMR üzerindeki ClickHouse'a yazmak için tam olarak bir kez (exactly-once) semantiği yalnızca VVR 4.0.11 ve sonraki sürümlerde kullanılabilir. Ancak ürün yeteneklerindeki değişiklikler nedeniyle bu semantik, ClickHouse on EMR v3.45.1 ve v5.11.1'den sonraki sürümler için kullanılamaz.

• Verileri yerel tablo düğümleri arasında eşit şekilde dağıtan balance yazma modu yalnızca VVR 8.0.7 ve sonraki sürümlerde kullanılabilir.

• ClickHouse yerel tabloya yazma işlemi yalnızca ApsaraDB for ClickHouse Community uyumlu Sürüm için desteklenir.

• Bir ClickHouse kataloğu kaydederken varsayılan veritabanı adı tire (-) içeremez. Aksi takdirde JDBC URL doğrulaması başarısız olur.

Söz dizimi

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

WITH parametreleri

Veri türü eşlemesi

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

Örnekler

• Örnek 1: Tek düğümlü bir tabloya yazma.

  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;

• Örnek 2: Dağıtılmış bir tabloya yazma.

  192.XX.XX.1, 192.XX.XX.2 ve 192.XX.XX.3 düğümlerinde bulunan local_table_test adlı üç yerel tablodan oluşturulan distributed_table_test adlı bir dağıtılmış tablo olduğunu varsayalım.

  • Flink'in verileri doğrudan yerel tablolara yazmasını ve verileri bir anahtara göre bölmesini istiyorsanız aşağıdaki DDL'yi kullanın:

    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;

  • Flink'in yerel tablo düğümlerini URL'de manuel olarak belirtmek yerine otomatik olarak keşfetmesini istiyorsanız aşağıdaki DDL'yi kullanın:

    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', -- The JDBC URL of a node in the cluster.
      'userName' = '<yourUsername>',
      'password' = '<yourPassword>',
      'tableName' = 'distributed_table_test', -- The name of the distributed table.
      'shardWrite' = 'true',
      'inferLocalTable' = 'true', -- Set inferLocalTable to true.
      'writeMode' = 'partition',
      'shardingKey' = 'name'
    );
    
    INSERT INTO clickhouse_output
    SELECT
      id,
      name,
      age,
      rate
    FROM clickhouse_source;

SSS

• ApsaraDB for ClickHouse sonuç tablosundan güncellenen veriler geri alınabilir mi?

• ClickHouse sink tablolarında veri görünürlüğü