Realtime Compute for Apache Flink:Conector do ClickHouse

Informações de contexto

O ClickHouse é um sistema de gerenciamento de banco de dados orientado a colunas para processamento analítico online (OLAP). Para mais informações, consulte O que é o ClickHouse?.

Recursos

• Grave dados diretamente nas tabelas locais de uma tabela distribuída do ClickHouse.

• Obtenha a semântica exactly-once ao gravar no ClickHouse no Alibaba Cloud E-MapReduce (EMR).

Pré-requisitos

• Crie uma tabela do ClickHouse. Para mais informações, consulte Criar uma nova tabela.

• Configure uma lista de permissões.

  • Se você usar o ApsaraDB for ClickHouse, consulte Configure a whitelist.

  • Se você usar o ClickHouse no Alibaba Cloud E-MapReduce (EMR), consulte Manage security groups.

  • Se você usar um cluster do ClickHouse autogerenciado em uma instância do ECS, consulte Security group overview.

  • Para todas as outras configurações, configure a lista de permissões na máquina onde o ClickHouse está implantado para permitir o acesso a partir da implantação do Realtime Compute for Apache Flink.

  Nota

  Para visualizar o segmento de rede do vSwitch para o Realtime Compute for Apache Flink, consulte How do I configure a whitelist?.

Limitações

• O parâmetro sink.parallelism não é suportado.

• Por padrão, o sink do ClickHouse fornece a semântica at-least-once.

• O conector do ClickHouse é suportado apenas no Ververica Runtime (VVR) 3.0.2 e versões posteriores.

• A opção ignoreDelete é suportada apenas no VVR 3.0.3, VVR 4.0.7 e versões posteriores.

• O tipo de dado Nested do ClickHouse é suportado apenas no VVR 4.0.10 e versões posteriores.

• Gravações diretas nas tabelas locais de uma tabela distribuída são suportadas apenas no VVR 4.0.11 e versões posteriores.

• A semântica exactly-once para gravação no ClickHouse no EMR está disponível apenas no VVR 4.0.11 e versões posteriores. No entanto, devido a alterações nas capacidades do product EMR, a semântica exactly-once também não está disponível para o ClickHouse no EMR em versões posteriores a 3.45.1 e 5.11.1.

• O modo de gravação balance, que distribui dados uniformemente entre os nós da tabela local, está disponível apenas no VVR 8.0.7 e versões posteriores.

• A gravação em uma tabela local do ClickHouse é suportada apenas para o ApsaraDB for ClickHouse Community-compatible Edition.

• Ao registrar um catálogo do ClickHouse, o nome do banco de dados padrão não pode conter um hífen (-). Caso contrário, a validação da URL JDBC falhará.

Sintaxe

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

Mapeamento de tipos de dados

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

Exemplos

• Exemplo 1: Grave em uma tabela de nó ú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;

• Exemplo 2: Grave em uma tabela distribuída.

  Suponha que uma tabela distribuída chamada distributed_table_test seja criada a partir de três tabelas locais chamadas local_table_test, localizadas nos nós 192.XX.XX.1, 192.XX.XX.2 e 192.XX.XX.3.

  • Se você quiser que o Flink grave dados diretamente nas tabelas locais e particione os dados por uma chave, use a seguinte 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;

  • Se você quiser que o Flink descubra automaticamente os nós da tabela local em vez de especificá-los manualmente na URL, use a seguinte 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;

Perguntas frequentes

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

• Data visibility in ClickHouse sink tables