向 Kafka 发送消息

登录 Kafka 客户端，执行如下命令，向 Kafka 发送消息，模拟网站点击流数据。

./bin/kafka-console-producer.sh --broker-list 10.1.0.10:9092 --topic uvpv-demo
{"record_type":0, "user_id": 2, "client_ip": "100.0.0.2", "product_id": 101, "create_time": "2021-09-08 16:20:00"}
{"record_type":0, "user_id": 3, "client_ip": "100.0.0.3", "product_id": 103, "create_time": "2021-09-08 16:21:00"}
{"record_type":1, "user_id": 2, "client_ip": "100.0.0.1", "product_id": 101, "create_time": "2021-09-08 16:22:10"}
{"record_type":0, "user_id": 2, "client_ip": "100.0.0.2", "product_id": 102, "create_time": "2021-09-08 16:23:20"}
{"record_type":0, "user_id": 3, "client_ip": "100.0.0.3", "product_id": 101, "create_time": "2021-09-08 16:24:50"}
{"record_type":1, "user_id": 1, "client_ip": "100.0.0.1", "product_id": 101, "create_time": "2021-09-08 16:25:00"}
{"record_type":0, "user_id": 1, "client_ip": "100.0.0.2", "product_id": 104, "create_time": "2021-09-08 16:27:00"}
{"record_type":0, "user_id": 3, "client_ip": "100.0.0.3", "product_id": 102, "create_time": "2021-09-08 16:29:00"}
{"record_type":1, "user_id": 2, "client_ip": "100.0.0.1", "product_id": 103, "create_time": "2021-09-08 16:30:00"}
{"record_type":0, "user_id": 2, "client_ip": "100.0.0.2", "product_id": 101, "create_time": "2021-09-08 16:31:00"}
{"record_type":0, "user_id": 3, "client_ip": "100.0.0.3", "product_id": 102, "create_time": "2021-09-08 16:32:00"}
{"record_type":1, "user_id": 2, "client_ip": "100.0.0.1", "product_id": 101, "create_time": "2021-09-08 17:33:00"}

验证结果

1. 通过 ClickHouse 连接工具登录 ClickHouse。

2. 执行以下命令，清空表数据。

   truncate table output_conversion_rate;
   truncate table output_pv;
   truncate table output_uv;
   

3. 执行以下命令，查询合并的数据。

   当前累计的 uv

   select userids,uv from output_uv final ;
   

   最新时刻的转化率

   select conversion_rate,rate from output_conversion_rate final ;
   

   每 10 分钟统计的 pv

   select sum(pv) as pv,stt,edt from output_pv group by stt,edt;
   

4. 或者也可以手动合并，正常查询。

   OPTIMIZE TABLE output_uv FINAL;
   OPTIMIZE TABLE output_conversion_rate FINAL;
   select * from output_uv;
   select * from output_conversion_rate;
   select sum(pv) as pv,stt,edt from output_pv group by stt,edt;