ApsaraDB for ClickHouse:Sincronizar dados do Kafka

Limitações

A sincronização de dados é compatível apenas com instâncias do ApsaraMQ for Kafka e clusters Kafka autogerenciados implantados em instâncias ECS.

Pré-requisitos

• ApsaraDB for ClickHouse:

  • Crie um cluster de destino na mesma região e VPC da instância do ApsaraMQ for Kafka. Para mais informações, consulte Create a cluster.

  • Crie uma conta de banco de dados com as permissões necessárias para o cluster de destino. Para mais informações, consulte Account Management.

• ApsaraMQ for Kafka:

  • Crie um topic has been created.

  • Crie um consumer group has been created.

Observações de uso

• O tópico assinado pela tabela externa Kafka do ApsaraDB for ClickHouse não deve ter outros consumers.

• Ao criar a tabela externa Kafka, a materialized view e a tabela local, garanta que os tipos de campo das três tabelas correspondam.

Procedimento

O exemplo a seguir sincroniza dados do ApsaraMQ for Kafka para a tabela distribuída kafka_table_distributed no banco de dados padrão de um cluster Community-compatible Edition do ApsaraDB for ClickHouse.

Etapa 1: Entender o funcionamento da sincronização

O ApsaraDB for ClickHouse usa o mecanismo de tabela Kafka e uma materialized view para consumir e armazenar dados do Kafka em tempo real. O fluxo de dados ocorre da seguinte forma:

• Tópico Kafka: contém os dados de origem a sincronizar.

• Tabela externa Kafka do ApsaraDB for ClickHouse (usa o mecanismo Kafka): extrai dados de origem de um tópico Kafka específico.

• Materialized view: lê dados de origem da tabela externa Kafka e os insere em uma tabela local no ApsaraDB for ClickHouse.

• Tabela local: armazena os dados sincronizados.

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'host:port1,host:port2,host:port3',
    kafka_topic_list = 'topic_name1,topic_name2,...',
    kafka_group_name = 'group_name',
    kafka_format = 'data_format'[,]
    [kafka_row_delimiter = 'delimiter_symbol',]
    [kafka_num_consumers = N,]
    [kafka_thread_per_consumer = 1,]
    [kafka_max_block_size = 0,]
    [kafka_skip_broken_messages = N,]
    [kafka_commit_every_batch = 0,]
    [kafka_auto_offset_reset = N]

CREATE TABLE default.kafka_src_table ON CLUSTER `default`
(
    -- Define the fields of the table schema.
    id Int32,
    name String               
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'alikafka-post-cn-****-1-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-****1-2-vpc.alikafka.aliyuncs.com:9092,alikafka-post-cn-****-3-vpc.alikafka.aliyuncs.com:9092',
    kafka_topic_list = 'testforCK',
    kafka_group_name = 'GroupForTestCK',
    kafka_format = 'CSV';

CREATE TABLE default.kafka_table_local ON CLUSTER default (
  id Int32,
  name String
) ENGINE = MergeTree()
ORDER BY (id);

CREATE MATERIALIZED VIEW <view_name> ON CLUSTER default TO <dest_table> AS SELECT * FROM <src_table>;

CREATE MATERIALIZED VIEW consumer ON CLUSTER default TO kafka_table_local AS SELECT * FROM kafka_src_table;

CREATE MATERIALIZED VIEW consumer ON CLUSTER default TO kafka_table_distributed AS SELECT * FROM kafka_src_table;

Etapa 6: Verificar a sincronização

1. Envie mensagens para o tópico na instância do ApsaraMQ for Kafka.

   1. Faça login no console do ApsaraMQ for Kafka.

   2. Na página Instance list, clique em nome da instância de destino.

   3. Na página Topics, localize o tópico de destino e selecione More > Send Message (Demo) na coluna Actions.

   4. Na página Send and Consume Message with Quick Experience, insira o Message Content.

      Este exemplo envia as mensagens 1,a e 2,b.

   5. Clique em OK.

2. Conecte-se ao cluster do ApsaraDB for ClickHouse, consulte a tabela distribuída e verifique se os dados foram sincronizados.

   Para fazer login em um cluster do ApsaraDB for ClickHouse, consulte Connect to an ApsaraDB for ClickHouse cluster by using DMS.

   Use as seguintes instruções para consultar e verificar os dados:

   Enterprise edition

   SELECT * FROM kafka_table_local; 

   Community-compatible edition

   Exemplo de consulta a uma tabela distribuída:

   • Se a tabela de destino for local, substitua o nome da tabela distribuída na consulta pelo nome da tabela local.

   • Em um cluster Community-compatible Edition com múltiplos nós, consulte a tabela distribuída. Consultar diretamente uma tabela local retorna dados de apenas um nó, resultando em um conjunto de resultados incompleto.

   SELECT * FROM kafka_table_distributed; 

   Se a consulta retornar resultados, a sincronização de dados do Kafka para o ApsaraDB for ClickHouse foi bem-sucedida.

   Os resultados da consulta são os seguintes.

   ┌─id─┬─name─┐
   │  1 │  a   │
   │  2 │  b   │
   └────┴──────┘

   Se os resultados da consulta não corresponderem ao esperado, prossiga para Step 7 (Optional): Check the consumption status of the Kafka external table para solucionar o problema.

select * from system.kafka_consumers;

SELECT * FROM system.kafka;

Perguntas frequentes

• Why are external tables for sources such as MySQL, HDFS, and Kafka unreachable?

• Why does the row count in ClickHouse not match the number of messages imported from Kafka?

• Why is no new data written to the Kafka external table after the table is created?