惠州企业建站系统,建设网站那家公司好,兰州生活网,广州互助网站开发文章目录 0 前言1、环境准备1.1 flink 下载相关 jar 包1.2 生成 kafka 数据1.3 开发前的三个小 tip 2、flink-sql 客户端编写运行 sql2.1 创建 kafka 数据源表2.2 指标统计#xff1a;每小时成交量2.2.1 创建 es 结果表#xff0c; 存放每小时的成交量2.2.2 执行 sql #x… 文章目录 0 前言1、环境准备1.1 flink 下载相关 jar 包1.2 生成 kafka 数据1.3 开发前的三个小 tip 2、flink-sql 客户端编写运行 sql2.1 创建 kafka 数据源表2.2 指标统计每小时成交量2.2.1 创建 es 结果表 存放每小时的成交量2.2.2 执行 sql 统计每小时的成交量 2.3 指标统计每10分钟累计独立用户数2.3.1 创建 es 结果表存放每10分钟累计独立用户数2.3.2 创建视图2.3.3 执行 sql 统计每10分钟的累计独立用户数 2.4 指标统计商品类目销量排行2.4.1 创建商品类目维表2.4.1 创建 es 结果表存放商品类目排行表2.4.2 创建视图2.4.3 执行 sql , 统计商品类目销量排行 3、最终效果与体验心得3.1 最终效果3.2 体验心得3.2.1 执行3.2.2 存储 4 最后 0 前言 这两年开始毕业设计和毕业答辩的要求和难度不断提升传统的毕设题目缺少创新和亮点往往达不到毕业答辩的要求这两年不断有学弟学妹告诉学长自己做的项目系统达不到老师的要求。
为了大家能够顺利以及最少的精力通过毕设学长分享优质毕业设计项目今天要分享的是 flink大数据淘宝用户行为数据实时分析与可视化
学长这里给一个题目综合评分(每项满分5分)
难度系数3分工作量3分创新点4分
1、环境准备
1.1 flink 下载相关 jar 包
flink-sql 连接外部系统时需要依赖特定的 jar 包所以需要事先把这些 jar 包准备好。说明与下载入口
本项目使用到了以下的 jar 包 下载后直接放在了 flink/lib 里面。 需要注意的是 flink-sql 执行时是转化为 flink-job 提交到集群执行的所以 flink 集群的每一台机器都要添加以下的 jar 包。 外部版本jarkafka4.1flink-sql-connector-kafka_2.11-1.10.2.jar flink-json-1.10.2-sql-jar.jarelasticsearch7.6flink-sql-connector-elasticsearch7_2.11-1.10.2.jarmysql5.7flink-jdbc_2.11-1.10.2.jar mysql-connector-java-8.0.11.jar
1.2 生成 kafka 数据
用户行为数据来源 阿里云天池公开数据集 网盘https://pan.baidu.com/s/1wDVQpRV7giIlLJJgRZAInQ 提取码gja5 商品类目纬度数据来源: category.sql
数据生成器datagen.py 有了数据文件之后使用 python 读取文件数据然后并发写入到 kafka。 修改生成器中的 kafka 地址配置然后运行 以下命令开始不断往 kafka 写数据 # 5000 并发
nohup python3 datagen.py 5000 1.3 开发前的三个小 tip 生成器往 kafka 写数据会自动创建主题无需事先创建 flink 往 elasticsearch 写数据会自动创建索引无需事先创建 Kibana 使用索引模式从 Elasticsearch 索引中检索数据以实现诸如可视化等功能。 使用的逻辑为创建索引模式 》Discover (发现) 查看索引数据 》visualize可视化创建可视化图表》dashboards仪表板创建大屏即汇总多个可视化的图表 2、flink-sql 客户端编写运行 sql
# 进入 flink-sql 客户端, 需要指定刚刚下载的 jar 包目录
./bin/sql-client.sh embedded -l lib2.1 创建 kafka 数据源表
-- 创建 kafka 表, 读取 kafka 数据
CREATE TABLE user_behavior (user_id BIGINT,item_id BIGINT,category_id BIGINT,behavior STRING,ts TIMESTAMP(3),proctime as PROCTIME(),WATERMARK FOR ts as ts - INTERVAL 5 SECOND
) WITH (connector.type kafka, connector.version universal, connector.topic user_behavior, connector.startup-mode earliest-offset, connector.properties.zookeeper.connect 172.16.122.24:2181, connector.properties.bootstrap.servers 172.16.122.17:9092, format.type json
);
SELECT * FROM user_behavior;2.2 指标统计每小时成交量
2.2.1 创建 es 结果表 存放每小时的成交量
CREATE TABLE buy_cnt_per_hour (hour_of_day BIGINT,buy_cnt BIGINT
) WITH (connector.type elasticsearch, connector.version 7, connector.hosts http://172.16.122.13:9200, connector.index buy_cnt_per_hour,connector.document-type user_behavior,connector.bulk-flush.max-actions 1,update-mode append,format.type json
);2.2.2 执行 sql 统计每小时的成交量
INSERT INTO buy_cnt_per_hour
SELECT HOUR(TUMBLE_START(ts, INTERVAL 1 HOUR)), COUNT(*)
FROM user_behavior
WHERE behavior buy
GROUP BY TUMBLE(ts, INTERVAL 1 HOUR);2.3 指标统计每10分钟累计独立用户数
2.3.1 创建 es 结果表存放每10分钟累计独立用户数
CREATE TABLE cumulative_uv (time_str STRING,uv BIGINT
) WITH (connector.type elasticsearch, connector.version 7, connector.hosts http://172.16.122.13:9200, connector.index cumulative_uv,connector.document-type user_behavior, update-mode upsert,format.type json
);2.3.2 创建视图
CREATE VIEW uv_per_10min AS
SELECTMAX(SUBSTR(DATE_FORMAT(ts, HH:mm),1,4) || 0) OVER w AS time_str,COUNT(DISTINCT user_id) OVER w AS uv
FROM user_behavior
WINDOW w AS (ORDER BY proctime ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW);2.3.3 执行 sql 统计每10分钟的累计独立用户数
INSERT INTO cumulative_uv
SELECT time_str, MAX(uv)
FROM uv_per_10min
GROUP BY time_str;2.4 指标统计商品类目销量排行
2.4.1 创建商品类目维表
先在 mysql 创建一张商品类目的维表然后配置 flink 读取 mysql。
CREATE TABLE category_dim (sub_category_id BIGINT,parent_category_name STRING
) WITH (connector.type jdbc,connector.url jdbc:mysql://172.16.122.25:3306/flink,connector.table category,connector.driver com.mysql.jdbc.Driver,connector.username root,connector.password root,connector.lookup.cache.max-rows 5000,connector.lookup.cache.ttl 10min
);2.4.1 创建 es 结果表存放商品类目排行表
CREATE TABLE top_category (category_name STRING,buy_cnt BIGINT
) WITH (connector.type elasticsearch, connector.version 7, connector.hosts http://172.16.122.13:9200, connector.index top_category,connector.document-type user_behavior,update-mode upsert,format.type json
);2.4.2 创建视图
CREATE VIEW rich_user_behavior AS
SELECT U.user_id, U.item_id, U.behavior, C.parent_category_name as category_name
FROM user_behavior AS U LEFT JOIN category_dim FOR SYSTEM_TIME AS OF U.proctime AS C
ON U.category_id C.sub_category_id;2.4.3 执行 sql , 统计商品类目销量排行
INSERT INTO top_category
SELECT category_name, COUNT(*) buy_cnt
FROM rich_user_behavior
WHERE behavior buy
GROUP BY category_name;3、最终效果与体验心得
3.1 最终效果
整个开发过程只用到了 flink-sql 无需写 java 或者其它代码就完成了这样一个实时报表。 3.2 体验心得
3.2.1 执行 flink-sql 的 ddl 语句不会触发 flink-job , 同时创建的表、视图仅在会话级别有效。 对于连接表的 insert、select 等操作则会触发相应的流 job 并自动提交到 flink 集群无限地运行下去直到主动取消或者 job 报错。 flink-sql 客户端关闭后对于已经提交到 flink 集群的 job 不会有任何影响。 本次开发执行了 3 个 insert , 因此打开 flink 集群面板可以看到有 3 个无限的流 job 。即使 kafka 数据全部写入完毕关闭 flink-sql 客户端这个 3 个 job 都不会停止。 3.2.2 存储 flnik 本身不存储业务数据只作为流批一体的引擎存在所以主要的用法为读取外部系统的数据处理后再写到外部系统。 flink 本身的元数据包括表、函数等默认情况下只是存放在内存里面所以仅会话级别有效。但是似乎可以存储到 Hive Metastore 中关于这一点就留到以后再实践。
4 最后
