Realtime Compute for Apache Flink:Conector de ClickHouse

Contexto

ClickHouse es un sistema de gestión de bases de datos orientado a columnas diseñado para el procesamiento analítico en línea (OLAP). Para obtener más información, consulte ¿Qué es ClickHouse?.

Funciones

• Escritura directa de datos en las tablas locales de una tabla distribuida de ClickHouse.

• Garantía de semántica exactly-once al escribir en ClickHouse en Alibaba Cloud E-MapReduce (EMR).

Requisitos previos

• Cree una tabla de ClickHouse. Para obtener más información, consulte Crear una nueva tabla.

• Configure una lista de permitidos.

  • Si utiliza ApsaraDB for ClickHouse, consulte Configure a whitelist.

  • Si utiliza ClickHouse en Alibaba Cloud E-MapReduce (EMR), consulte Manage security groups.

  • Si utiliza un clúster de ClickHouse autogestionado en una instancia ECS, consulte Security group overview.

  • Para cualquier otra configuración, configure la lista de permitidos en la máquina donde se despliega ClickHouse para permitir el acceso desde el despliegue de Realtime Compute for Apache Flink.

  Nota

  Para consultar el segmento de red del vSwitch de Realtime Compute for Apache Flink, consulte How do I configure a whitelist?.

Limitaciones

• El parámetro sink.parallelism no es compatible.

• De forma predeterminada, el sink de ClickHouse ofrece semántica at-least-once.

• El conector de ClickHouse solo es compatible con Ververica Runtime (VVR) 3.0.2 y versiones posteriores.

• La opción ignoreDelete solo es compatible con VVR 3.0.3, VVR 4.0.7 y versiones posteriores.

• El tipo de datos Nested de ClickHouse solo es compatible con VVR 4.0.10 y versiones posteriores.

• La escritura directa en las tablas locales de una tabla distribuida solo es compatible con VVR 4.0.11 y versiones posteriores.

• La semántica exactly-once para escribir en ClickHouse en EMR solo está disponible en VVR 4.0.11 y versiones posteriores. No obstante, debido a cambios en las capacidades del producto, esta semántica no está disponible para ClickHouse en EMR v3.45.1 ni en versiones posteriores a la v5.11.1.

• El modo de escritura balance, que distribuye los datos uniformemente entre los nodos de la tabla local, solo está disponible en VVR 8.0.7 y versiones posteriores.

• La escritura en una tabla local de ClickHouse solo es compatible con ApsaraDB for ClickHouse Community-compatible Edition.

• Al registrar un catálogo de ClickHouse, el nombre de la base de datos predeterminada no puede contener un guion (-). De lo contrario, la validación de la URL JDBC fallará.

Sintaxis

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

Parámetros WITH

Mapeo de tipos de datos

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

Ejemplos

• Ejemplo 1: Escritura en una tabla de nodo único.

  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;

• Ejemplo 2: Escritura en una tabla distribuida.

  Suponga que se crea una tabla distribuida llamada distributed_table_test a partir de tres tablas locales llamadas local_table_test, ubicadas en los nodos 192.XX.XX.1, 192.XX.XX.2 y 192.XX.XX.3.

  • Si desea que Flink escriba datos directamente en las tablas locales y particione los datos mediante una clave, utilice la siguiente DDL:

    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;

  • Si desea que Flink descubra automáticamente los nodos de la tabla local en lugar de especificarlos manualmente en la URL, utilice la siguiente DDL:

    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;

Preguntas frecuentes

• Can I retract the updated data from an ApsaraDB for ClickHouse result table?

• Data visibility in ClickHouse sink tables