【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;