-- 0 ckafka 查询导入到 ClickHouse 的 topic 信息
-- 1 在 ClickHouse 中创建 Kafka 引擎表
CREATE TABLE kafka_table_qiang_test_1 ON CLUSTER default_cluster (
    key String,
    value String
) ENGINE = Kafka
SETTINGS
    kafka_broker_list = '10.0.x.x:9092',
    kafka_topic_list = 'qiang-test-mysql',
    kafka_group_name = 'cg_qiang-test-mysql_1',
    kafka_format = 'JSONEachRow',  -- 可以根据你的数据格式选择合适的格式，例如 JSON、Avro、CSV
    kafka_num_consumers = 1,  -- 按需调整
    kafka_max_block_size = 65536,
    kafka_skip_broken_messages = 0; --默认是0

-- 2 创建目标表用于存储数据
--Kafka 引擎表本身只消费数据，不持久化，需要创建一个实际存储数据的表
CREATE TABLE real_table_qiang_test_1 ON CLUSTER default_cluster (
    key String,
    value String
) ENGINE = MergeTree()
ORDER BY key;

-- 3 创建物化视图（Materialized View）将数据从 Kafka 表导入到实际表中
CREATE MATERIALIZED VIEW view_to_real_table_qiang_test_1 ON CLUSTER default_cluster  TO real_table_qiang_test_1 AS
SELECT
    key,
    value
FROM kafka_table_qiang_test_1;

-- 4 当开始往kafka发送数据的时候，查询 ClickHouse 中的数据
SELECT * FROM real_table_qiang_test_1;
select count(1) from INFORMATION_SCHEMA.real_table_qiang_test_1 limit 5; 


-- 5 若要暂停物化视图 来停止数据采集
DROP VIEW IF EXISTS view_to_real_table_qiang_test_1 ON CLUSTER default_cluster;

SELECT name FROM system.tables WHERE name LIKE '%view_to_real_table_qiang_test_1%'; -- 验证删除成功