ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink学习连载文章13--FlinkSQL高级部门 [打印本页]
作者: 欢乐狗    时间: 2024-12-11 11:14
标题: Flink学习连载文章13--FlinkSQL高级部门
eventTime

测试数据如下：
  1. {"username":"zs","price":20,"event_time":"2023-07-17 10:10:10"}
  2. {"username":"zs","price":15,"event_time":"2023-07-17 10:10:30"}
  3. {"username":"zs","price":20,"event_time":"2023-07-17 10:10:40"}
  4. {"username":"zs","price":20,"event_time":"2023-07-17 10:11:03"}
  5. {"username":"zs","price":20,"event_time":"2023-07-17 10:11:04"}
  6. {"username":"zs","price":20,"event_time":"2023-07-17 10:12:04"}
  7. {"username":"zs","price":20,"event_time":"2023-07-17 11:12:04"}
  8. {"username":"zs","price":20,"event_time":"2023-07-17 11:12:04"}
  9. {"username":"zs","price":20,"event_time":"2023-07-17 12:12:04"}
  10. {"username":"zs","price":20,"event_time":"2023-07-18 12:12:04"}
复制代码
需求：每隔1分钟统计这1分钟的每个用户的总消费金额和消费次数
需要用到滚动窗口

编写好sql:
  1. CREATE TABLE table1 (
  2.   `username` string,
  3.   `price` int,
  4.   `event_time` TIMESTAMP(3),
  5.   watermark for event_time as event_time - interval '3' second
  6. ) WITH (
  7.   'connector' = 'kafka',
  8.   'topic' = 'topic1',
  9.   'properties.bootstrap.servers' = 'bigdata01:9092',
  10.   'properties.group.id' = 'g1',
  11.   'scan.startup.mode' = 'latest-offset',
  12.   'format' = 'json'
  13. );
  15. 编写sql：
  16. select
  17.    window_start,
  18.    window_end,
  19.    username,
  20.    count(1) zongNum,
  21.    sum(price) totalMoney
  22.    from table(TUMBLE(TABLE table1, DESCRIPTOR(event_time), INTERVAL '60' second))
  23. group by window_start,window_end,username;
复制代码
分享一个错误：
Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. The window function TUMBLE(TABLE table_name, DESCRIPTOR(timecol), datetime interval) requires the timecol is a time attribute type, but is VARCHAR(2147483647).
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:156)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:107)

说明创建窗口的时候，使用的字段不是时间字段，需要写成时间字段TIMESTAMP(3)，使用了eventtime需要添加水印，否则报错。
需求：按照滚动窗口和EventTime进行统计，每隔1分钟统计每个人的消费总额是多少
  1. package com.bigdata.day08;
  3. import org.apache.flink.api.common.RuntimeExecutionMode;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. /**
  8. * @基本功能:
  9. * @program:FlinkDemo
  10. * @author: 闫哥
  11. * @create:2023-11-28 14:12:28
  12. **/
  13. public class _03EventTimeGunDongWindowDemo {
  15.     public static void main(String[] args) throws Exception {
  17.         //1. env-准备环境
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
  22.         //2. 创建表
  23.         tenv.executeSql("CREATE TABLE table1 (\n" +
  24.                         "  `username` String,\n" +
  25.                         "  `price` int,\n" +
  26.                         "  `event_time` TIMESTAMP(3),\n" +
  27.                         "   watermark for event_time as event_time - interval '3' second\n" +
  28.                         ") WITH (\n" +
  29.                         "  'connector' = 'kafka',\n" +
  30.                         "  'topic' = 'topic1',\n" +
  31.                         "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
  32.                         "  'properties.group.id' = 'testGroup1',\n" +
  33.                         "  'scan.startup.mode' = 'group-offsets',\n" +
  34.                         "  'format' = 'json'\n" +
  35.                         ")");
  36.         //3. 通过sql语句统计结果
  38.         tenv.executeSql("select \n" +
  39.                         "   window_start,\n" +
  40.                         "   window_end,\n" +
  41.                         "   username,\n" +
  42.                         "   count(1) zongNum,\n" +
  43.                         "   sum(price) totalMoney \n" +
  44.                         "   from table(TUMBLE(TABLE table1, DESCRIPTOR(event_time), INTERVAL '60' second))\n" +
  45.                         "group by window_start,window_end,username").print();
  46.         //4. sink-数据输出
  49.         //5. execute-执行
  50.         env.execute();
  51.     }
  52. }
复制代码
统计效果如下：


测试一下滑动窗口，每隔10秒钟，盘算前1分钟的数据：
  1. package com.bigdata.day08;
  3. import org.apache.flink.api.common.RuntimeExecutionMode;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. /**
  8. * @基本功能:
  9. * @program:FlinkDemo
  10. * @author: 闫哥
  11. * @create:2023-11-28 14:12:28
  12. **/
  13. public class _03EventTimeGunDongWindowDemo {
  15.     public static void main(String[] args) throws Exception {
  17.         //1. env-准备环境
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
  22.         //2. 创建表
  23.         tenv.executeSql("CREATE TABLE table1 (\n" +
  24.                 "  `username` String,\n" +
  25.                 "  `price` int,\n" +
  26.                 "  `event_time` TIMESTAMP(3),\n" +
  27.                 "   watermark for event_time as event_time - interval '3' second\n" +
  28.                 ") WITH (\n" +
  29.                 "  'connector' = 'kafka',\n" +
  30.                 "  'topic' = 'topic1',\n" +
  31.                 "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
  32.                 "  'properties.group.id' = 'testGroup1',\n" +
  33.                 "  'scan.startup.mode' = 'group-offsets',\n" +
  34.                 "  'format' = 'json'\n" +
  35.                 ")");
  36.         //3. 通过sql语句统计结果
  38.         tenv.executeSql("select \n" +
  39.                 "   window_start,\n" +
  40.                 "   window_end,\n" +
  41.                 "   username,\n" +
  42.                 "   count(1) zongNum,\n" +
  43.                 "   sum(price) totalMoney \n" +
  44.                 "   from table(HOP(TABLE table1, DESCRIPTOR(event_time), INTERVAL '10' second,INTERVAL '60' second))\n" +
  45.                 "group by window_start,window_end,username").print();
  46.         //4. sink-数据输出
  49.         //5. execute-执行
  50.         env.execute();
  51.     }
  52. }
复制代码
效果如图所示：




  1. package com.bigdata.day08;
  3. import org.apache.flink.api.common.RuntimeExecutionMode;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. /**
  8. * @基本功能:
  9. * @program:FlinkDemo
  10. * @author: 闫哥
  11. * @create:2023-11-28 14:12:28
  12. **/
  13. public class _03EventTimeGunDongWindowDemo {
  15.     public static void main(String[] args) throws Exception {
  17.         //1. env-准备环境
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.         StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
  22.         //2. 创建表
  23.         tenv.executeSql("CREATE TABLE table1 (\n" +
  24.                 "  `username` String,\n" +
  25.                 "  `price` int,\n" +
  26.                 "  `event_time` TIMESTAMP(3),\n" +
  27.                 "   watermark for event_time as event_time - interval '3' second\n" +
  28.                 ") WITH (\n" +
  29.                 "  'connector' = 'kafka',\n" +
  30.                 "  'topic' = 'topic1',\n" +
  31.                 "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
  32.                 "  'properties.group.id' = 'testGroup1',\n" +
  33.                 "  'scan.startup.mode' = 'group-offsets',\n" +
  34.                 "  'format' = 'json'\n" +
  35.                 ")");
  36.         //3. 通过sql语句统计结果
  38.         tenv.executeSql("select \n" +
  39.                 "   window_start,\n" +
  40.                 "   window_end,\n" +
  41.                 "   username,\n" +
  42.                 "   count(1) zongNum,\n" +
  43.                 "   sum(price) totalMoney \n" +
  44.                 "   from table(CUMULATE(TABLE table1, DESCRIPTOR(event_time), INTERVAL '1' hours,INTERVAL '1' days))\n" +
  45.                 "group by window_start,window_end,username").print();
  46.         //4. sink-数据输出
  49.         //5. execute-执行
  50.         env.execute();
  51.     }
  52. }
复制代码
累积窗口演示效果：


processTime

测试数据：
  1. {"username":"zs","price":20}
  2. {"username":"lisi","price":15}
  3. {"username":"lisi","price":20}
  4. {"username":"zs","price":20}
  5. {"username":"zs","price":20}
  6. {"username":"zs","price":20}
  7. {"username":"zs","price":20}
复制代码
  1. /**
  2. * 滚动窗口大小1分钟 延迟时间3秒
  3. *
  4. * {"username":"zs","price":20}
  5. * {"username":"lisi","price":15}
  6. * {"username":"lisi","price":20}
  7. * {"username":"zs","price":20}
  8. * {"username":"zs","price":20}
  9. * {"username":"zs","price":20}
  10. * {"username":"zs","price":20}
  11. *
  12. */
  13. package com.bigdata.day08;
  15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  16. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  18. /**
  19. * @基本功能:
  20. * @program:FlinkDemo
  21. * @author: 闫哥
  22. * @create:2023-11-28 14:12:28
  23. **/
  24. public class _04ProcessingTimeGunDongWindowDemo {
  26.     public static void main(String[] args) throws Exception {
  28.         //1. env-准备环境
  29.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  30.         env.setParallelism(1);
  31.         StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
  33.         //2. 创建表
  34.         tenv.executeSql("CREATE TABLE table1 (\n" +
  35.                 "  `username` String,\n" +
  36.                 "  `price` int,\n" +
  37.                 "  `event_time` as proctime()\n" +
  38.                 ") WITH (\n" +
  39.                 "  'connector' = 'kafka',\n" +
  40.                 "  'topic' = 'topic1',\n" +
  41.                 "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
  42.                 "  'properties.group.id' = 'testGroup1',\n" +
  43.                 "  'scan.startup.mode' = 'group-offsets',\n" +
  44.                 "  'format' = 'json'\n" +
  45.                 ")");
  46.         //3. 通过sql语句统计结果
  48.         tenv.executeSql("select \n" +
  49.                 "   window_start,\n" +
  50.                 "   window_end,\n" +
  51.                 "   username,\n" +
  52.                 "   count(1) zongNum,\n" +
  53.                 "   sum(price) totalMoney \n" +
  54.                 "   from table(TUMBLE(TABLE table1, DESCRIPTOR(event_time), INTERVAL '60' second ))\n" +
  55.                 "group by window_start,window_end,username").print();
  56.         //4. sink-数据输出
  59.         //5. execute-执行
  60.         env.execute();
  61.     }
  62. }
复制代码
盘算效果：


效果需要等1分钟，才能表现出来，不要着急！
窗口分为滚动和滑动，时间分为事件时间和处置惩罚时间，两两组合，4个案例。
以下是滑动窗口+处置惩罚时间：
  1. package com.bigdata.sql;
  3. import org.apache.flink.api.common.RuntimeExecutionMode;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. /**
  8. * @基本功能:
  9. * @program:FlinkDemo
  10. * @author: 闫哥
  11. * @create:2024-11-29 14:28:19
  12. **/
  13. public class _04_FlinkSQLProcessTime_HOP {
  15.     public static void main(String[] args) throws Exception {
  17.         //1. env-准备环境
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
  20.         // 获取tableEnv对象
  21.         // 通过env 获取一个table 环境
  22.         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
  24.         tEnv.executeSql("CREATE TABLE table1 (\n" +
  25.                         "  `username` string,\n" +
  26.                         "  `price` int,\n" +
  27.                         "  `event_time` as proctime() \n"+
  28.                         ") WITH (\n" +
  29.                         "  'connector' = 'kafka',\n" +
  30.                         "  'topic' = 'topic1',\n" +
  31.                         "  'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
  32.                         "  'properties.group.id' = 'g1',\n" +
  33.                         "  'scan.startup.mode' = 'latest-offset',\n" +
  34.                         "  'format' = 'json'\n" +
  35.                         ")");
  37.         // 语句中的 ; 不能添加
  38.         tEnv.executeSql("select \n" +
  39.                         "   window_start,\n" +
  40.                         "   window_end,\n" +
  41.                         "   username,\n" +
  42.                         "   count(1) zongNum,\n" +
  43.                         "   sum(price) totalMoney \n" +
  44.                         "   from table(HOP(TABLE table1, DESCRIPTOR(event_time),INTERVAL '10' second, INTERVAL '60' second))\n" +
  45.                         "group by window_start,window_end,username").print();
  48.         //5. execute-执行
  49.         env.execute();
  50.     }
  51. }
复制代码


测试时假如你的控制台不出数据，触发不了，请进入如下操作：
1、重新创建一个新的 topic，分区数为 1
2、kafka 对接的 server，写全 bigdata01:9092,bigdata02:9092,bigdata03:9092
二、窗口TopN(不是新的技术)

需求：在每个小时内找出点击量最多的Top 3网页。
  1. 测试数据
  2. {"ts": "2023-09-05 12:00:00", "page_id": 1, "clicks": 100}
  3. {"ts": "2023-09-05 12:01:00", "page_id": 2, "clicks": 90}
  4. {"ts": "2023-09-05 12:10:00", "page_id": 3, "clicks": 110}
  5. {"ts": "2023-09-05 12:20:00", "page_id": 4, "clicks": 23}
  6. {"ts": "2023-09-05 12:30:00", "page_id": 5, "clicks": 456}
  7. {"ts": "2023-09-05 13:10:00", "page_id": 5, "clicks": 456}
复制代码
  1. 假如没有每隔1小时的需求，仅仅是统计点击量最多的Top 3网页，结果如下
  2. select * from (
  3. select
  4.     page_id,
  5.     totalSum,
  6.     row_number() over (order by totalSum desc) px
  7.   from (
  8.      select page_id,
  9.       sum(clicks)  totalSum
  10.       from kafka_page_clicks group by page_id )  ) where px <=3;
复制代码
根据以上代码，添加滚动窗口的写法：
  1. select
  2.     window_start,
  3.     window_end,
  4.     page_id,
  5.     sum(clicks) totalSum  
  6.     from
  7.    table (
  8.      tumble( table kafka_page_clicks, descriptor(ts), INTERVAL '1' HOUR )
  9.          )
  10.     group by window_start,window_end,page_id;
  13. 在这个基础之上添加排名的写法：
  14. select
  15.    window_start,
  16.    window_end,
  17.    page_id,
  18.    pm
  19.   from   (
  20. select
  21.     window_start,
  22.     window_end,
  23.     page_id,
  24.     row_number() over(partition by window_start,window_end order by totalSum desc ) pm
  25.   from (
  26. select
  27.     window_start,
  28.     window_end,
  29.     page_id,
  30.     sum(clicks) totalSum  
  31.     from
  32.    table (
  33.      tumble( table kafka_page_clicks, descriptor(ts), INTERVAL '1' HOUR )
  34.          )
  35.     group by window_start,window_end,page_id ) t2 ) t1  where pm <= 3;
复制代码
编写建表语句：
  1. {"ts": "2023-09-05 12:00:00", "page_id": 1, "clicks": 100}
  3. CREATE TABLE kafka_page_clicks (
  4.   `ts` TIMESTAMP(3),
  5.   `page_id` int,
  6.   `clicks` int,
  7.   watermark for ts as ts - interval '3' second
  8. ) WITH (
  9.   'connector' = 'kafka',
  10.   'topic' = 'topic1',
  11.   'properties.bootstrap.servers' = 'bigdata01:9092',
  12.   'properties.group.id' = 'g1',
  13.   'scan.startup.mode' = 'latest-offset',
  14.   'format' = 'json'
  15. )
复制代码
  1. package com.bigdata.day08;
  3. import org.apache.flink.api.common.RuntimeExecutionMode;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. /**
  8. * @基本功能:
  9. * @program:FlinkDemo
  10. * @author: 闫哥
  11. * @create:2023-11-28 15:23:46
  12. **/
  13. public class _05TopNDemo {
  15.     public static void main(String[] args) throws Exception {
  17.         //1. env-准备环境
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.         // ctrl + y 删除光标所在的那一行数据  ctrl + d 复制当前行
  21.         StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
  23.         //2. source-加载数据
  24.         // 一定要注意：ts 是一个年月日时分秒的数据，所以在建表时一定要是TIMESTAMP，否则进行WATERMARK 报错
  25.         // 因为使用的是event_time 所以，需要指定WATERMARK
  26.         tenv.executeSql("CREATE TABLE kafka_page_clicks (" +
  27.                 "    `ts` TIMESTAMP(3),\n" +
  28.                 "    page_id INT,\n" +
  29.                 "    clicks INT,\n" +
  30.                 "  WATERMARK FOR ts AS ts - INTERVAL '10' SECOND \n" +
  31.                 ") WITH (\n" +
  32.                 "    'connector' = 'kafka',\n" +
  33.                 "    'topic' = 'topic1',\n" +
  34.                 "    'properties.bootstrap.servers' = 'bigdata01:9092',\n" +
  35.                 "   'scan.startup.mode' = 'group-offsets',\n" +
  36.                 "    'format' = 'json'\n" +
  37.                 ")");
  40.         tenv.executeSql("select \n" +
  41.                 "   window_start,\n" +
  42.                 "   window_end,\n" +
  43.                 "   page_id,\n" +
  44.                 "   pm\n" +
  45.                 "  from   (\n" +
  46.                 "select \n" +
  47.                 "    window_start,\n" +
  48.                 "    window_end,\n" +
  49.                 "    page_id,\n" +
  50.                 "    row_number() over(partition by window_start,window_end order by totalSum desc ) pm\n" +
  51.                 "  from (\n" +
  52.                 "select \n" +
  53.                 "    window_start,\n" +
  54.                 "    window_end,\n" +
  55.                 "    page_id,\n" +
  56.                 "    sum(clicks) totalSum  \n" +
  57.                 "    from \n" +
  58.                 "   table ( \n" +
  59.                 "     tumble( table kafka_page_clicks, descriptor(ts), INTERVAL '1' HOUR ) \n" +
  60.                 "         ) \n" +
  61.                 "    group by window_start,window_end,page_id ) t2 ) t1  where pm <= 3").print();
  62.         //4. sink-数据输出
  65.         //5. execute-执行
  66.         env.execute();
  67.     }
  68. }
复制代码
最后的运行效果如下：



免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4