Realtime Compute for Apache Flink:Connecteur ClickHouse

Informations générales

ClickHouse est un système de gestion de base de données orienté colonne destiné au traitement analytique en ligne (OLAP). Pour plus d'informations, consultez la documentation Qu'est-ce que ClickHouse ?.

Fonctionnalités

• Écriture directe des données dans les tables locales d'une table distribuée ClickHouse.

• Garantie de la sémantique exactly-once lors de l'écriture dans ClickHouse sur Alibaba Cloud E-MapReduce (EMR).

Prérequis

• Créez une table ClickHouse. Pour plus d'informations, consultez Créer une nouvelle table.

• Configurez une liste d'autorisation.

  • Si vous utilisez ApsaraDB for ClickHouse, consultez Configurer une liste d'autorisation.

  • Si vous utilisez ClickHouse sur Alibaba Cloud E-MapReduce (EMR), consultez Gérer les groupes de sécurité.

  • Si vous utilisez un cluster ClickHouse auto-géré sur une instance ECS, consultez Présentation des groupes de sécurité.

  • Pour toutes les autres configurations, configurez la liste d'autorisation sur la machine où ClickHouse est déployé afin d'autoriser l'accès depuis le déploiement Realtime Compute for Apache Flink.

  Remarque

  Pour afficher le segment de réseau vSwitch pour Realtime Compute for Apache Flink, consultez Comment configurer une liste d'autorisation ?.

Limitations

• Le paramètre sink.parallelism n'est pas pris en charge.

• Par défaut, le puits ClickHouse offre une sémantique at-least-once.

• Le connecteur ClickHouse est pris en charge uniquement dans Ververica Runtime (VVR) 3.0.2 et versions ultérieures.

• L'option ignoreDelete est prise en charge uniquement dans VVR 3.0.3, VVR 4.0.7 et versions ultérieures.

• Le type de données Nested de ClickHouse est pris en charge uniquement dans VVR 4.0.10 et versions ultérieures.

• Les écritures directes dans les tables locales d'une table distribuée sont prises en charge uniquement dans VVR 4.0.11 et versions ultérieures.

• La sémantique exactly-once pour l'écriture dans ClickHouse sur EMR est disponible uniquement dans VVR 4.0.11 et versions ultérieures. Toutefois, en raison de modifications des capacités du produit, cette sémantique n'est pas disponible pour ClickHouse sur EMR v3.45.1 et les versions supérieures à v5.11.1.

• Le mode d'écriture balance, qui répartit uniformément les données sur les nœuds de table locale, est disponible uniquement dans VVR 8.0.7 et versions ultérieures.

• L'écriture dans une table locale ClickHouse est prise en charge uniquement pour l'édition compatible avec ApsaraDB for ClickHouse Community.

• Lorsque vous enregistrez un catalogue ClickHouse, le nom de la base de données par défaut ne peut pas contenir de trait d'union (-). Dans le cas contraire, la validation de l'URL JDBC échoue.

Syntaxe

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

Paramètres WITH

Mappage des types de données

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

Exemples

• Exemple 1 : Écriture dans une table à nœud unique.

  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;

• Exemple 2 : Écriture dans une table distribuée.

  Supposons qu'une table distribuée nommée distributed_table_test soit créée à partir de trois tables locales nommées local_table_test, situées sur les nœuds 192.XX.XX.1, 192.XX.XX.2 et 192.XX.XX.3.

  • Si vous souhaitez que Flink écrive les données directement dans les tables locales et partitionne les données par clé, utilisez la DDL suivante :

    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 vous souhaitez que Flink découvre automatiquement les nœuds de table locale au lieu de les spécifier manuellement dans l'URL, utilisez la DDL suivante :

    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;

FAQ

• Puis-je rétracter les données mises à jour d'une table de résultats ApsaraDB for ClickHouse ?

• Visibilité des données dans les tables puits ClickHouse