我应该如何将 clickhouse 连接到 Kafka？答案

【问题标题】：How should I connect clickhouse to Kafka?我应该如何将 clickhouse 连接到 Kafka？
【发布时间】：2021-07-27 13:24:55
【问题描述】：

CREATE TABLE readings_queue
(

    `readid` Int32,

    `time` DateTime,

    `temperature` Decimal(5,2)
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'serverIP:9092',
 kafka_topic_list = 'newtest',
 kafka_format = 'CSV',
 kafka_group_name = 'clickhouse_consumer_group',
 kafka_num_consumers = 3

上面是我建立 Kafka 和 Clickhouse 连接的代码。但是我执行代码后，只创建了表，没有取回任何数据。

~/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic newtest --from-beginning
1,"2020-05-16 23:55:44",14.2
2,"2020-05-16 23:55:45",20.1
3,"2020-05-16 23:55:51",12.9

我在 clickhouse 中的查询有问题吗？

当我检查日志时，我发现以下警告。

2021.05.10 10:19:50.534441 [ 1534 ] {} <Warning> (readings_queue): Can't get assignment. It can be caused by some issue with consumer group (not enough partitions?). Will keep trying.

【问题讨论】：

不清楚是什么将数据生成到该表或主题中
看这篇文章-altinity.com/blog/clickhouse-kafka-engine-faq
+ 尝试添加参数 kafka_row_delimiter = '\n'
@OneCricketeer 数据应该用'，'分隔。例如第一行，readid:1, time:"2020-05-16 23:55:44", temp:14.2.
@OneCricketeer 在 clickhouse 中构建表格后，我尝试为主题生成新数据。但它也不起作用。

标签： apache-kafka clickhouse

【解决方案1】：

这是一个可以使用的示例：

-- Actual table to store the data fetched from an Apache Kafka topic
CREATE TABLE data(
  a DateTime,
  b Int64,
  c Sting
) Engine=MergeTree ORDER BY (a,b);

-- Kafka Engine which consumes the data from 'data_topic' of Apache Kafka
CREATE TABLE data_kafka(
  a DateTime,
  b Int64,
  c Sting
) ENGINE = Kafka
SETTINGS kafka_broker_list = '10.1.2.3:9092,10.1.2.4:9092,
10.1.2.5:9092',
 kafka_topic_list = 'data_topic'
 kafka_num_consumers = 1,
 kafka_group_name = 'ch-result5',
 kafka_format = 'JSONEachRow';

-- MV to insert any consumed data from Kafka Engine to 'data' table
CREATE MATERIALIZED VIEW data_mv TO data AS
SELECT
  a,
  b,
  c
FROM data_kafka;

【讨论】：