【kafka KSQL】游戏日志统计分析（3）

接上篇文章 【kafka KSQL】游戏日志统计分析（2），本文主要通过实例展示KSQL的连接查询功能。

创建另一个topic

bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 4 --topic propnew-normalized

往新topic中写入数据

bin/kafka-console-producer --broker-list localhost:9092 --topic propnew-normalized
>
{"user__name":"lzb", "prop__id":"id1"}

从prop-normalized主题创建Stream

CREATE STREAM PROP_USE_EVENT \
    (user__name VARCHAR, \
     prop__id VARCHAR ) \
     WITH (KAFKA_TOPIC='propnew-normalized', \
           VALUE_FORMAT='json');

重新设置ROWKEY为user__name

CREATE STREAM PROP_USE_EVENT_REKEY AS \
    SELECT * FROM PROP_USE_EVENT \
    PARTITION BY user__name;

查询完成3局对局且没有使用过道具的所有玩家

• 查询出所有玩家的对局情况，并创建表USER_SCORE_TABLE(前面已经创建过了)：

CREATE TABLE USER_SCORE_TABLE AS \
    SELECT username, COUNT(*) AS game_count, SUM(delta) AS delta_sum, SUM(tax) AS tax_sum \
    FROM USER_SCORE_EVENT_REKEY \
    WHERE reason = 'game' \
    GROUP BY username;

• 查询出所有玩家的道具使用情况，并创建表USER_PROP_TABLE：

CREATE TABLE USER_PROP_TABLE AS \
    SELECT user__name, COUNT(*) AS use_count \
    FROM PROP_USE_EVENT_REKEY \
    GROUP BY user__name;

• 使用LEFT JOIN进行左关联，并以此创建一个新的TABLE：

CREATE TABLE USER_SCORE_AND_PROP AS \
SELECT s.username AS username, s.game_count, s.tax_sum, s.delta_sum, p.use_count \
FROM USER_SCORE_TABLE s \
LEFT JOIN USER_PROP_TABLE p \
ON s.username = p.user__name;

• 查询对局数大于等于3，且没有使用过道具的玩家：

SELECT username FROM USER_SCORE_AND_PROP \
WHERE game_count >= 3 AND use_count IS NULL;

• 查询对局数大于等于3，且使用道具次数大于等于2的玩家：

SELECT username FROM USER_SCORE_AND_PROP \
WHERE game_count >= 3 AND use_count >= 2;