使用FlinkSql实现热门商品TOP N

Dear 丶 2022-11-07 14:55 414阅读 0赞

# 使用FlinkSql实现热门商品TOP N #

目前仅 Blink 计划器支持 Top-N 。

Flink 使用`OVER 窗口条件`和`过滤条件`相结合以进行 Top-N 查询。利用 OVER 窗口的 PARTITION BY 子句的功能，Flink 还支持逐组 Top-N 。 例如，每个类别中实时销量最高的前五种产品。批处理表和流处理表都支持基于SQL的 Top-N 查询。

`流处理模式需注意`: TopN 查询的结果会带有更新。 Flink SQL 会根据排序键对输入的流进行排序；若 top N 的记录发生了变化，变化的部分会以撤销、更新记录的形式发送到下游。 `推荐使用一个支持更新的存储作为 Top-N 查询的 sink`。另外，若 top N 记录需要存储到外部存储，则结果表需要拥有与 Top-N 查询相同的`唯一键`。

## 1. 需求描述 ##

每隔10min 统计最近 1hour的热门商品 top3, 并把统计的结果写入到mysql中

思路:

1.  统计每个商品的点击量, 开窗
2.  分组窗口分组,
3.  over窗口

## 2. 在mysql中创建表 ##

CREATE DATABASE flink_sql;
    USE flink_sql;
    DROP TABLE IF EXISTS `hot_item`;
    CREATE TABLE `hot_item` (
      `w_end` timestamp NOT NULL,
      `item_id` bigint(20) NOT NULL,
      `item_count` bigint(20) NOT NULL,
      `rk` bigint(20) NOT NULL,
      PRIMARY KEY (`w_end`,`rk`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

## 3. 导入JDBC Connector依赖 ##

<properties>
            <flink.version>1.12.0</flink.version>
            <scala.binary.version>2.11</scala.binary.version>
            <slf4j.version>1.7.30</slf4j.version>
     </properties>
    
    <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-java</artifactId>
                <version>${flink.version}</version>
                
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>5.1.49</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-csv</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-json</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-jdbc_2.11</artifactId>
                <version>1.12.0</version>
            </dependency>
        </dependencies>

## 4. 具体实现代码 ##

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
    
    public class Flink01_HotItem_TopN { 
        public static void main(String[] args) { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(2);
            StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
    
    
            // 使用sql从文件读取数据
            tenv.executeSql(
                "create table user_behavior(" +
                    " user_id bigint, " +
                    " item_id bigint, " +
                    " category_id int, " +
                    " behavior string, " +
                    " ts bigint, " +
                    " event_time as to_timestamp(from_unixtime(ts, 'yyyy-MM-dd HH:mm:ss')), " +
                    " watermark for event_time as event_time - interval '5' second " +
                    ")with(" +
                    " 'connector'='filesystem', " +
                    " 'path'='input/UserBehavior.csv', " +
                    " 'format'='csv')"
            );
    
            // 每隔 10m 统计一次最近 1h 的热门商品 top
    
            // 1. 计算每每个窗口内每个商品的点击量
            Table t1 = tenv
                .sqlQuery(
                    "select " +
                        " item_id, " +
                        " hop_end(event_time, interval '10' minute, interval '1' hour) w_end," +
                        " count(*) item_count " +
                        "from user_behavior " +
                        "where behavior='pv' " +
                        "group by hop(event_time, interval '10' minute, interval '1' hour), item_id"
                );
            tenv.createTemporaryView("t1", t1);
            // 2. 按照窗口开窗, 对商品点击量进行排名
            Table t2 = tenv.sqlQuery(
                "select " +
                    " *," +
                    " row_number() over(partition by w_end order by item_count desc) rk " +
                    "from t1"
            );
            tenv.createTemporaryView("t2", t2);
    
            // 3. 取 top3
            Table t3 = tenv.sqlQuery(
                "select " +
                    " item_id, w_end, item_count, rk " +
                    "from t2 " +
                    "where rk<=3"
            );
    
            // 4. 数据写入到mysql
            // 4.1 创建输出表
            tenv.executeSql("create table hot_item(" +
                                " item_id bigint, " +
                                " w_end timestamp(3), " +
                                " item_count bigint, " +
                                " rk bigint, " +
                                " PRIMARY KEY (w_end, rk) NOT ENFORCED)" +
                                "with(" +
                                " 'connector' = 'jdbc', " +
                                " 'url' = 'jdbc:mysql://hadoop162:3306/flink_sql?useSSL=false', " +
                                " 'table-name' = 'hot_item', " +
                                " 'username' = 'root', " +
                                " 'password' = 'aaaaaa' " +
                                ")");
            // 4.2 写入到输出表
            t3.executeInsert("hot_item");
        }
    }