热搜榜TOP10，一小时区间内每五分钟刷新

package com.ctgu.flink.project;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.types.Row;


public class Flink_Sql_HotItem {
    public static void main(String[] args) throws Exception {

        long start = System.currentTimeMillis();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 注册函数
        tableEnv.createFunction("toTs", TimestampFunction.class);

        String createSql =
                "CREATE TABLE source " +
                "    (" +
                "    `userId` BIGINT," +
                "    `itemId` BIGINT," +
                "    `categoryId` INT," +
                "    `behavior` STRING," +
                "    `ts` BIGINT" +
                "    )" +
                "    WITH (" +
                "       'connector'='filesystem'," +
                "       'format'='csv'," +
                "       'csv.field-delimiter'=','," +
                "       'path'='data/UserBehavior.csv'" +
                "    )";

        tableEnv.executeSql(createSql);

        String userBehavior = "select *, ts * 1000 as `timestamp` from source where behavior = 'pv'";

        Table userBehaviorTable = tableEnv.sqlQuery(userBehavior);

        DataStream<Row> rowDataStream = tableEnv.toDataStream(userBehaviorTable);

        Table source =
                tableEnv.fromDataStream(
                        rowDataStream,
                        Schema.newBuilder()
                                .columnByExpression("time_ltz", "TO_TIMESTAMP_LTZ(`timestamp`, 3)")
                                .watermark("time_ltz", "time_ltz - INTERVAL '5' SECOND")
                                .build());

        tableEnv.createTemporaryView("userBehavior", source);

        String hotItemSql =
                "       SELECT window_start, window_end, itemId, COUNT(*) as cnt" +
                        "  FROM TABLE(" +
                        "    HOP(Table userBehavior, DESCRIPTOR(time_ltz), INTERVAL '5' minutes, INTERVAL '1' hours)) " +
                        "  GROUP BY window_start, window_end, itemId";

        String top10Sql =
                "       SELECT *" +
                        "  FROM (" +
                        "    SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY cnt DESC) as rownum" +
                        "    FROM (" +
                        "      SELECT window_start, window_end, itemId, COUNT(*) as cnt" +
                        "      FROM TABLE(" +
                        "        HOP(Table userBehavior, DESCRIPTOR(time_ltz), INTERVAL '5' minutes, INTERVAL '1' hours)) " +
                        "      GROUP BY window_start, window_end, itemId" +
                        "    )" +
                        "  ) WHERE rownum <= 10";

        Table top10 = tableEnv.sqlQuery(top10Sql);

        tableEnv.toDataStream(top10, Row.class).print("top10");

        env.execute("Table SQL");

        System.out.println("耗时: " + (System.currentTimeMillis() - start) / 1000);
    }

    public static class TimestampFunction extends ScalarFunction {

        public Long eval(Long value) {
            return value * 1000L;
        }
    }

}

测试数据 UserBehavior.csv

543462,1715,1464116,pv,1511658000
662867,2244074,1575622,pv,1511658000
561558,3611281,965809,pv,1511658000
894923,3076029,1879194,pv,1511658000
834377,4541270,3738615,pv,1511658000
315321,942195,4339722,pv,1511658000
625915,1162383,570735,pv,1511658000
578814,176722,982926,pv,1511658000
873335,1256540,1451783,pv,1511658000
429984,4625350,2355072,pv,1511658000
866796,534083,4203730,pv,1511658000
937166,321683,2355072,pv,1511658000
156905,2901727,3001296,pv,1511658000
758810,5109495,1575622,pv,1511658000
107304,111477,4173315,pv,1511658000
452437,3255022,5099474,pv,1511658000
813974,1332724,2520771,buy,1511658000
524395,3887779,2366905,pv,1511658000
470572,3760258,1299190,pv,1511658001
543789,3110556,4558987,cart,1511658001
354759,2191348,4756105,pv,1511658001
382009,2123538,4801426,pv,1511658001
677046,1598945,4145813,pv,1511658001
946161,3021357,1506018,pv,1511658001
464646,2512167,2733371,pv,1511658001
1007641,5046581,2355072,pv,1511658001
723938,4719377,1464116,pv,1511658001
513008,3472922,401357,pv,1511658001
769215,22738,2355072,pv,1511658002
652863,4967749,1320293,pv,1511658002
801610,900305,634390,pv,1511658002
411478,3259235,2667323,pv,1511658002
431664,764155,2520377,pv,1511658002
487768,4125503,2465336,pv,1511658002
223813,4104826,2042400,pv,1511658002
672849,1822977,4801426,fav,1511658002
550127,4602135,65362,pv,1511658002
205752,1467139,171529,pv,1511658002
64419,2029769,2729260,pv,1511658002
756093,2881426,2520377,pv,1511658002
48353,4362292,583014,pv,1511658002
355509,4712764,4082778,pv,1511658003
826492,4016552,2735466,pv,1511658003
624915,2243311,2520377,pv,1511658003
682317,655740,982926,fav,1511658003
677621,1051389,4801426,pv,1511658003
422974,4649255,4818107,pv,1511658003
86512,563566,4756105,pv,1511658003
565218,2331370,3607361,pv,1511658003
232313,4182588,1730376,pv,1511658003
436966,1329977,3607361,cart,1511658003
561158,269170,2342116,fav,1511658003
344379,3318242,2920476,cart,1511658003
858204,2450718,235534,pv,1511658004
833924,3190176,1051370,pv,1511658004
992993,1900968,3794706,fav,1511658004
911930,1150136,2131531,pv,1511658004
736959,319911,4756105,pv,1511658004
82170,3588374,2465336,pv,1511658004
587599,2067643,4818107,cart,1511658004
367451,15775,4756105,pv,1511658004
428316,2478780,4284875,pv,1511658004
284910,3680091,3829657,pv,1511658004
345119,737662,4357323,pv,1511658004
551442,1762997,1879194,pv,1511658004
550384,3908776,1029459,pv,1511658004
677500,4534693,2640118,pv,1511658004
398626,2791489,1467750,pv,1511658004
118053,3545571,2433095,pv,1511658005
457401,4063698,4801426,pv,1511658005
45105,3234847,3141941,fav,1511658005
604760,2661651,3738615,pv,1511658005
905383,2064903,2939262,cart,1511658005
740788,3657484,4936889,pv,1511658005
456838,1242724,4756105,fav,1511658005
585217,215764,2640118,pv,1511658006
658185,4025021,4048584,fav,1511658006
210431,2035568,2328673,pv,1511658006
602619,1838725,2247787,pv,1511658006
860388,3797303,4357323,pv,1511658006
175334,2624960,801221,pv,1511658006
72403,4249007,1320293,pv,1511658006
307385,2551880,4050612,pv,1511658006
819283,2094785,2520377,pv,1511658006
801272,565658,1158475,pv,1511658006
344680,3224461,4789432,pv,1511658006
125206,1102775,622168,pv,1511658006
59131,1960832,154040,pv,1511658006
252339,2455388,3745824,pv,1511658006
794780,4465604,4242717,pv,1511658007
388283,4701157,1457367,pv,1511658007
416261,2101120,1299190,pv,1511658007
231758,3622677,4758477,pv,1511658007
92253,642337,4135185,pv,1511658007
297958,1762578,4801426,pv,1511658007
786771,1940649,1320293,pv,1511658007
789048,3144191,2355072,pv,1511658007
895384,1138468,1602288,pv,1511658007
578800,1324176,4135836,pv,1511658007
886777,4606952,996587,pv,1511658008