Flink Sql介绍
在实际开发过程中，我个人是以Stream API为主，Flink Sql写的不多，这里主要是参考原本项目代码写的，具体的细节，我也不是很熟悉，建议大家可以单独去了解一下Flink Sql；简单来说就是就是先通过Sql语句从对应的数据表（这里主要就是DWM层的订单宽表）来获取所需要的字段数据，形成动态表，然后将动态表再转化为流的形式，再存到ClickHouse中去。
要注意的点：
1、Flink Sql和MySql的编写语言略有不同；
2、转化为流的时候，注意选用的模式，Flink中有三种模式：
将 Table 转换成 DataStream
追加模式（Append Mode）
– 用于表只会被插入（Insert）操作更改的场景
DataStream<Row> resultStream = tableEnv.toAppendStream(resultTable, Row.class);
撤回模式（Retract Mode）
– 用于任何场景。有些类似于更新模式中 Retract 模式，它只有 Insert 和 Delete 两类操作。
– 得到的数据会增加一个 Boolean 类型的标识位（返回的第一个字段），用它来表示到底是新增的数据（Insert），还是被删除的数据（Delete）
DataStream<Tuple2<Boolean, Row>> aggResultStream = tableEnv.toRetractStream(aggResultTable , Row.class);
地区主题宽表制作
这张表我不单独来讲，因为设计比较简单，具体的思路代码我在项目中都写得很清楚了，就是根据省市做分区统计，这里对应的就是项目流程一种的那张中国地图的热图分布；订单金额的总计就是大屏金额总计值；
关键词宽表制作
关键词的展示也是一种维度聚合的结果，根据聚合的大小来决定关键词的大小。
关键词的第一重要来源的就是用户在搜索栏的搜索，另外就是从以商品为主题的统计
中获取关键词。
//这里讲的是用户在搜索栏中的搜索关键词，从dwd_page_log中的"page_id":"good_list","item":"图书"中获取内容；
所以我们需要根据把长文本分割成一个一个的词，这种分词技术，在搜索引擎中可能会用到。对于中文分词，现在的搜索引擎基本上都是使用的第三方分词器，咱们在计
算数据中也可以，使用和搜索引擎中一致的分词器，IK。
//如何从Sql语句中去完成分词呢？这里用到自定义函数；
//案例：
//@FunctionHint(output = @DataTypeHint("ROW<s STRING, i INT>"))
public static class OverloadedFunction extends TableFunction<Row> {
public void eval(int a, int b) {
collect(Row.of("Sum", a + b));
}
// overloading of arguments is still possible
public void eval() {
collect(Row.of("Empty args", -1));
}
}
下面是UDTF函数定义时的一个注解说明：
//@FunctionHint(output = @DataTypeHint("ROW<word STRING>"))指定的就是输出的类型；
解耦类型推导与求值方法，类型推导完全取决于 FunctionHint
//@FunctionHint(
// input = {@DataTypeHint("INT"), @DataTypeHint("INT")},
// output = @DataTypeHint("INT")
//)
//流程：
//1.创建环境、注册自定义函数；（注意，这里用的是UDTF，从后续的的聚合思路展示中可以看到，是多行输出结果；
//2.创建动态表；
//3.从动态表查询数据；（where page['page_id']='good_list' and page['item'] IS NOT NULL）商品页面的搜索词；
//4.聚合；
//5.转换为流；
//6.写入clickhouse；
//clickhouse创建关键词主题宽表：
create table keyword_stats_0709 (
stt DateTime,
edt DateTime,
keyword String ,
source String ,
ct UInt64 ,
ts UInt64
)engine =ReplacingMergeTree( ts)
partition by toYYYYMMDD(stt)
order by ( stt,edt,keyword,source );
// select函数中常规的join用法：
SELECT *
FROM Orders
INNER JOIN Product
ON Orders.product_id = Product.id
select中表函数的join方法；
SELECT order_id, res
FROM Orders, //Orders是一个Table；
LATERAL TABLE(table_func(order_id)) t(res)
//LATERAL TABLE表示关联；
dwd_page_log数据：我们需要的是其中"page_id":"good_list","item":"图书",的内容；
page:2> {"common":{"ar":"530000","uid":"36","os":"Android 11.0","ch":"wandoujia","is_new":"0",
"md":"Xiaomi 10 Pro ","mid":"mid_9","vc":"v2.1.134","ba":"Xiaomi"},"page":{"page_id":"good_list","item":"图书",
"during_time":7183,"item_type":"keyword","last_page_id":"search"},"displays":[{"display_type":"recommend",
"item":"1","item_type":"sku_id","pos_id":5,"order":1},{"display_type":"recommend","item":"5",
"item_type":"sku_id","pos_id":2,"order":2},{"display_type":"query","item":"2","item_type":"sku_id","pos_id":2,
"order":3},{"display_type":"promotion","item":"5","item_type":"sku_id","pos_id":3,"order":4},
{"display_type":"promotion","item":"9","item_type":"sku_id","pos_id":4,"order":5},{"display_type":"query",
"item":"9","item_type":"sku_id","pos_id":5,"order":6}],"ts":1626684732000}
//整体测试
? 启动 ZK、Kafka、logger.sh、ClickHouse
? 运行 BaseLogApp
? 运行 KeywordStatsApp
? 运行 rt_applog 目录下的 jar 包
? 查看控制台输出
? 查看 ClickHouse 中 keyword_stats_0709 表数据
//结果类型为：
// >>>>:4> KeywordStats(keyword=盒子, ct=2, source=SEARCH, stt=2021-07-19 12:58:30, edt=2021-07-19 12:58:40, ts=1626757129000)
//clickhouse中数据为：
────────────────stt─┬─────────────────edt─┬─keyword─┬─source─┬─ct─┬────────────ts─┐
│ 2021-07-19 13:00:50 │ 2021-07-19 13:01:00 │ 电视 │ SEARCH │ 1 │ 1626757293000 │
│ 2021-07-19 13:01:00 │ 2021-07-19 13:01:10 │ 图书 │ SEARCH │ 2 │ 1626757293000 │
│ 2021-07-19 13:01:00 │ 2021-07-19 13:01:10 │ 电视 │ SEARCH │ 1 │ 1626757293000 │
│ 2021-07-19 13:01:00 │ 2021-07-19 13:01:10 │ 盒子 │ SEARCH │ 1 │ 1626757293000 │
│ 2021-07-19 13:01:10 │ 2021-07-19 13:01:20 │ 电视 │ SEARCH │ 1 │ 1626757293000 │
//从以商品为主题的统计中获取关键词。
//从商品主题获得，商品关键词与点击次数、订单次数、添加购物次数的统计表。存入clickhouse表keyword_stats_0709中；
//聚合计算思路展示：
商品主题统计完毕之后，UDTF的设计思路图展示如下：
product_stats_0709
商品SPU 点击次数 下单次数 加购次数
小米手机 50 80 100
经过分词之后，得到的结果
关键词 点击次数 下单次数 加购次数
小米 50 80 100
手机 50 80 100
目标是统计：关键词热度--->Keyword_Stats_0820
关键词 操作次数 来源
小米 50 Click
小米 80 Order
小米 100 Cart
手机 50 Click
手机 80 Order
手机 100 Cart
获取热词 搜索 10 点击 9 下单 8 加购 7
关键词 操作次数 来源 评分
小米 90 搜索 10
小米 50 Click 9
小米 80 Order 8
小米 100 Cart 7
手机 50 Click
手机 80 Order
手机 100 Cart
sum
关键词 90 * 10
50 * 9
80 * 8
100 * 6
整体测试
? 启动 ZK、Kafka、logger.sh、ClickHouse、Redis、HDFS、Hbase、Maxwell
? 运行 BaseLogApp
? 运行 BaseDBApp
? 运行 OrderWideApp
? 运行 PaymentWideApp
? 运行 ProductsStatsApp
? 运行 KeywordStats4ProductApp
? 运行 rt_applog 目录下的 jar 包
? 运行 rt_dblog 目录下的 jar 包
? 查看控制台输出
? 查看 ClickHouse 中 products_stats_0709 表数据
keywordStatsProductDataStream数据类型为：
//分别对应点击、订单和购物车；
1> KeywordStats(keyword=10x, ct=2, source=CLICK, stt=2021-07-20 13:29:10, edt=2021-07-20 13:29:20, ts=1626758982000)
1> KeywordStats(keyword=redmi, ct=12, source=ORDER, stt=2021-07-20 13:28:50, edt=2021-07-20 13:29:00, ts=1626758978000)
1> KeywordStats(keyword=清, ct=178, source=CART, stt=2021-07-19 21:58:10, edt=2021-07-19 21:58:20, ts=1626758976000)
clickhouse中数据为：
┌─────────────────stt─┬─────────────────edt─┬─keyword───┬─source─┬─ct─┬────────────ts─┐
│ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ 10 │ CLICK │ 1 │ 1626758976000 │
│ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ 10x │ CLICK │ 1 │ 1626758977000 │
│ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ redmi │ CLICK │ 1 │ 1626758977000 │
│ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ 小米 │ CLICK │ 1 │ 1626758976000 │
│ 2021-07-20 13:27:50 │ 2021-07-20 13:28:00 │ 10 │ CLICK │ 2 │ 1626758977000 │
│ 2021-07-20 13:27:50 │ 2021-07-20 13:28:00 │ 10 │ ORDER │ 9 │ 1626758977000 │
//购物订单数据比较少，所以这里不好找，但是确实是有三种数据的；
总结
整个项目到这里就基本结束了，后续的可视化环节以及数据接口的编写，我这里不展开讲解，因为涉及到了SpringBoot的一个简单使用，有兴趣大家也可以自行查看代码，都是一些简单的数据统计流程，然后按照sugar要求的格式去转化为json格式；有兴趣的话，各位可以去研究一下，技术不难，都是苦力活和细致活。
对这个项目或者Flink技术感兴趣的话也可以继续关注我，后续我也会开始逐步更新Flink框架以及实时和流式处理的一些学习方法和路线，希望大家一起进步~~
还是把我最后总结的数据接口的笔记思路放在这里，仅供参考：
接口访问路径以及返回格式
? 访问路径
/api/sugar/gmv
? 返回格式
{
"status": 0,
"msg": "",
"data": 1201081.1632389291
}
6.2数据接口实现
6.2.1总成交金额接口
//数据来源：product_stats_0709
启动zk,kafka,logger.sh(nginx + 日志处理服务),maxwell,hdfs,hbase,Redis,ClichHouse
BaseLogApp,BaseDBApp,OrderWideApp,PaymentWide,ProductStatsApp
//web访问的顺序：
date --》 controller --》 service --》 mapper --》 service --》controller；
//测试流程：
1) 启动 SpringBoot 应用程序
2) 用浏览器访问测试接口
//http://localhost:8070/api/sugar/gmv；
0.实现方式
-使用SpringBoot开发数据服务接口
-使用的技术
Spring SpringMVC Mybatis
-对web项目进行了分层 Controller<-->Service<-->Mapper 面向接口编程、面向抽象编程不要面向具体
*表示层 Controller
和客户端打交道，接收用户的情况，将请求交给业务层处理，返回响应给客户端
*业务层 Service
处理具体的业务
*持久层 dao|mapper
和数据库打交道
1.总成交金额
-组件
数字翻牌器
-请求地址
$API_HOST/api/sugar/gmv
-返回数据的格式
{
"status": 0,
"msg": "",
"data": 1201076.1961842624
}
-执行SQL
select sum(order_amount) from product_stats_0709 where toYYYYMMDD(stt)=20210721
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
6.2.2商品交易额不同维度的统计
//数据来源：product_stats_0709
启动zk,kafka,logger.sh(nginx + 日志处理服务),maxwell,hdfs,hbase,Redis,ClichHouse
BaseLogApp,BaseDBApp,OrderWideApp,PaymentWide,ProductStatsApp
三个关于商品交易额方面的统计
? 品牌，水平柱状图
? 品类，饼形图
? 商品 spu，轮播图
这三个的共同特征是可以根据商品统计信息计算出来。
2.按品牌统计销售排行
-组件
横向柱状图
-请求地址
$API_HOST/api/sugar/trademark?limit=5
-返回数据的格式
{
"status": 0,
"data": {
"categories": ["苹果","三星","华为"],
"series": [
{
"data": [
9387,
8095,
8863
]
}
]
}
}
-执行SQL
select tm_id,tm_name,sum(order_amount) order_amount from product_stats0709 where toYYYYMMDD(stt)=#{date} group by tm_id,tm_name having order_amount >0 order by order_amount desc limit #{limit}
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
3.按品类统计销售排行
-组件
轮播饼图
-请求地址
$API_HOST/api/sugar/category3?date=20210720&limit=3
-返回数据的格式
//data中的数据对应的就是饼图中的类型和对应值；
{
"status": 0,
"data": [
{
"name": "PC",
"value": 97
},
{
"name": "iOS",
"value": 50
}
]
}
-执行SQL
select category3_id,category3_name,sum(order_amount) order_amount from product_stats_0709 where toYYYYMMDD(stt)=20210721 group by category3_id,category3_name having order_amount >0 order by order_amount desc limit 10;
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4.SPU销售排行
-组件
轮播表格
-请求地址
$API_HOST/api/sugar/spu?limit=10
-返回数据的格式
// * columns其实就是轮播表格中的第一行数据，其name就是展示的值，id就是对应的数据值；
// * rows对应的就是实际每列的数据；数组中一个对象就是一行数据的对应值；
{
"status": 0,
"data": {
"columns": [
{
"name": "分组织",
"id": "subsys"
}
],
"rows": [
{
"subsys": "北京总部"
}
]
}
}
-执行SQL
select spu_id,spu_name,sum(order_amount) order_amount,sum(order_ct) order_ct from product_stats_0709 where toYYYYMMDD(stt)=20210721 group by spu_id,spu_name having order_amount >0 order by order_amount desc limit 10;
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
6.2.3地区统计交易额
//得到province_stats_0709数据：
? 启动 ZK、Kafka、ClickHouse、Redis、HDFS、Hbase、Maxwell
? 运行 BaseDBApp
? 运行 OrderWideApp
? 运行 ProvinceStatsSqlApp
? 运行 rt_dblog 目录下的 jar 包
5.按地区统计交易额
-组件
中国省份色彩
-请求地址
$API_HOST/api/sugar/province
-返回数据的格式
//只需要省份名字和订单总额两个数值即可；
{
"status": 0,
"data": {
"mapData": [
{
"name": "北京",
"value": 7489
}
]
}
}
-执行SQL
select province_name,sum(order_amount) order_amount from province_stats_0709 where toYYYYMMDD(stt)=#{date} group by province_id,province_name
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
6.2.4新老用户对比
//访客实时统计、用户分时统计 得到visitor_stats_0709数据；
* 需要启动的服务
* -log.sh(Nginx以及日志处理服务)、zk、kafka
* -BaseLogApp、UniqueVisitApp、UserJumpDetailApp、VisitorStatsApp
SELECT *
FROM visitor_stats_0709
┌─────────────────stt─┬─────────────────edt─┬─vc───────┬─ch────────┬─ar─────┬─is_new─┬─uv_ct─┬─pv_ct─┬─sv_ct─┬─uj_ct─┬─dur_sum─┬────────────ts─┐
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.0.1 │ wandoujia │ 110000 │ 0 │ 0 │ 2 │ 1 │ 0 │ 14602 │ 1626663133462 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.0.1 │ xiaomi │ 110000 │ 0 │ 1 │ 9 │ 1 │ 0 │ 114957 │ 1626663133462 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.132 │ Appstore │ 110000 │ 0 │ 1 │ 2 │ 1 │ 0 │ 21965 │ 1626663133414 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.134 │ Appstore │ 110000 │ 0 │ 0 │ 9 │ 1 │ 0 │ 89691 │ 1626663133462 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.134 │ Appstore │ 530000 │ 0 │ 1 │ 9 │ 1 │ 0 │ 102535 │ 1626663133415 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.134 │ oppo │ 310000 │ 0 │ 2 │ 4 │ 2 │ 0 │ 56512 │ 1626663133415 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.134 │ oppo │ 370000 │ 0 │ 1 │ 7 │ 1 │ 0 │ 83905 │ 1626663133462 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.134 │ wandoujia │ 110000 │ 0 │ 0 │ 2 │ 1 │ 0 │ 36541 │ 1626663133415 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.134 │ wandoujia │ 370000 │ 0 │ 1 │ 6 │ 1 │ 0 │ 84847 │ 1626663133461 │
│ 2021-07-19 10:52:00 │ 2021-07-19 10:52:10 │ v2.1.134 │ xiaomi │ 440000 │ 0 │ 1 │ 6 │ 1 │ 0 │ 93489 │ 1626663133414 │
6.新老访客访问指标对比
-组件
表格
-请求地址
$API_HOST/api/sugar/visitor?date=...
-返回数据的格式
//跳出率：跳出次数/访问次数（这里用sv而不是uv和pv）；（也即会话数）
//平均在线时长：
{
"status": 0,
"data": {
"combineNum": 1,
"columns": [
{
"name": "类别",
"id": "type"
},
{
"name": "新用户",
"id": "new"
},
{
"name": "老用户",
"id": "old"
}
],
"rows": [
{
"type": "用户数",
"new": 123,
"old": 13
},
{
"type": "总访问页面",
"new": 123,
"old": 145
},
{
"type": "跳出率",
"new": 123,
"old": 145
},
{
"type": "平均在线时长",
"new": 123,
"old": 145
},
{
"type": "平均访问页面数",
"new": 23,
"old": 145
}
]
}
}
-执行SQL
select is_new,sum(uv_ct) uv_ct,sum(pv_ct) pv_ct,sum(sv_ct) sv_ct, sum(uj_ct) uj_ct,sum(dur_sum) dur_sum from visitor_stats_0709 where toYYYYMMDD(stt)=20210719 group by is_new
//这里都是sum后的结果，group by的列又只有0和1两个数据，所以查询出来的结果就只有两个结果;
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
6.2.5用户分时统计
//访客实时统计、用户分时统计 得到visitor_stats_0709数据；
* 需要启动的服务
* -log.sh(Nginx以及日志处理服务)、zk、kafka
* -BaseLogApp、UniqueVisitApp、UserJumpDetailApp、VisitorStatsApp
7.访客分时统计排行
-组件
折线图
-请求地址
$API_HOST/api/sugar/hr?date=20210721
//visitor_stats_0709访客主题宽表中获取数据；
-返回数据的格式
{
// //categories表示的是横轴字段；
//series表示的是
"status": 0,
"data": {
"categories": [
"00",
"01",
"02"
],
"series": [
{
"name": "UV",
"data": [
2838,
3872,
1388
]
},
{
"name": "PV",
"data": [
2838,
3872,,
1388
]
},
{
"name": "新访客",
"data": [
2838,
3872,,
1388
]
}
]
}
}
-执行SQL
//uv和pv直接取sum值即可；
//newUv的计算方法，找到uv_ct为1的值的数据中,is_new也为1的数据，把这些数据数目统计出来；
select
sum(if(is_new='1', visitor_stats_0709.uv_ct,0)) new_uv,
toHour(stt) hr,
sum(visitor_stats_0709.uv_ct) uv_ct,
sum(pv_ct) pv_ct,
sum(uj_ct) uj_ct
from visitor_stats_0709 where toYYYYMMDD(stt)=20210721 group by toHour(stt) order by toHour(stt)
//结果为：
─new_uv─┬─hr─┬─uv_ct─┬─uj_ct─┐
│ 19 │ 16 │ 60 │ 0 │
│ 0 │ 17 │ 0 │ 0
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
6.2.6关键词词云
//数据来源：keyword_stats_0709
? 启动 ZK、Kafka、logger.sh、ClickHouse
? 运行 BaseLogApp
? 运行 KeywordStatsApp
? 运行 rt_applog 目录下的 jar 包
热词字符云
-组件
3D词云
-请求地址
$API_HOST/api/sugar/keyword?date=20210721?limit=5
-返回数据的格式
{
"status": 0,
"data": [
{
"name": "visualMap",
"value": 22199
},
{
"name": "brush",
"value": 3
}
]
}
-执行SQL
select
keyword,
sum(
keyword_stats_0709.ct *
multiIf(
source='SEARCH',10,source='ORDER',5,source='CART',2,source='CLICK',1,0
)
) ct
from
keyword_stats_0709
where
toYYYYMMDD(stt) = 20210721
group by
keyword
order by
ct;
----------------------------------------------------