写点什么

Flink 实例:电商用户行为实时分析

作者:TiAmo
  • 2023-06-06
    江苏
  • 本文字数:5574 字

    阅读完需:约 18 分钟

Flink实例:电商用户行为实时分析

【示例 1】综合运用 Flink 的各种 API,基于 EventTime 实现分析电商用户行为。

电商平台中的用户行为频繁且较复杂,系统上线运行一段时间后,可以收集到大量的用户行为数据,进而利用大数据技术进行深入挖掘和分析,得到感兴趣的商业指标并增强对风险的控制。

电商用户行为数据多样,整体可以分为用户行为习惯数据和业务行为数据两大类。用户的行为习惯数据包括了用户的登录方式、上线的时间点及时长、单击和浏览页面、页面停留时间及页面跳转等,可以从中进行流量统计和热门商品的统计,也可以深入挖掘用户的特征;这些数据往往可以从 web 服务器日志中直接读取,而业务行为数据就是用户在电商平台中针对每个业务(通常是某个具体商品)所作的操作,一般会在业务系统中相应的位置埋点,然后收集日志进行分析。业务行为数据又可以简单分为两类:一类是能够明显地表现出用户兴趣的行为,例如对商品的收藏、喜欢、评分和评价,可以从中对数据进行深入分析,得到用户画像,进而对用户给出个性化的推荐商品列表,这个过程往往会用到机器学习相关的算法;另一类则是常规的业务操作,但需要着重关注一些异常状况以做好风控,例如登录和订单支付。

本项目限于数据,只实现实时热门商品统计。

01、数据源说明

本案例使用阿里天池的一份淘宝用户行为数据集,格式为 csv 文件。本数据集包含了 2023 年 1 月 25 日至 2023 年 5 月 3 日之间,约一百万随机用户的所有行为(行为包括单击、购买、加购、喜欢)。数据集的每行表示一条用户行为,由用户 ID、商品 ID、商品类目 ID、行为类型和时间戳组成,并以逗号分隔。

注意/


从天池上下载的数据集 UserBehavior.csv 解压缩后为 3.41G,这里截取其中一部分(485730 行)用于开发测试,命名为 UserBehavior_part.csv。

关于数据集中每列的详细描述,见表 1。

■ 表 1 淘宝用户行为数据集说明


其中用户行为类型共有 4 种,它们分别是:

(1) pv:商品详情页 pv,等价于单击。

(2) buy:商品购买。

(3) cart:将商品加入购物车。

(4) fav:收藏商品。

部分用户行为数据示例如下:

543462,1715,1464116,pv,1511658000662867,2244074,1575622,pv,1511658000561558,3611281,965809,pv,1511658000894923,3076029,1879194,pv,1511658000834377,4541270,3738615,pv,1511658000...
复制代码

02、预备知识

Flink SQL 提供了如下一些与日期处理有关的函数。

1)TO_TIMESTAMP

将 BIGINT 类型的日期或者 VARCHAR 类型的日期转换成 TIMESTAMP 类型,其语法如下:

TIMESTAMP TO_TIMESTAMP(BIGINT time) //time:毫秒TIMESTAMP TO_TIMESTAMP(VARCHAR date) //date:yyyy-MM-dd HH:mm:ssTIMESTAMP TO_TIMESTAMP(VARCHAR date, VARCHAR format)
复制代码


2)FROM_UNIXTIME

返回值为 VARCHAR 类型的日期值,默认日期格式:yyyy-MM-dd HH:mm:ss,若指定日期格式则按指定格式输出。如果任一输入参数是 NULL,返回 NULL,其语法如下:

VARCHAR FROM_UNIXTIME(BIGINT UNIXtime[, VARCHAR format])
复制代码

说明/


(1) 参数 UNIXtime 为长整型,是以秒为单位的时间戳。

(2) 参数 format 可选,为日期格式,默认格式为 yyyy-MM-dd HH:mm:ss,表示返回 VARCHAR 类型的符合指定格式的日期,如果有参数为 null 或解析错误,则返回 null。

03、任务实现


(1) 将用户行为数据采集到 Kafka。


(2) 使用 Table API 读取 Kafka 并写入 MySQL。


(3)用 Grafana 实时可视化显示。


首先,项目添加依赖。因为要读取的 Kafka 的用户行为事件是 CSV 格式的,所以在项目的 pom.xml 文件中添加如下的依赖:

<!-- 需要添加flink-csv依赖 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-csv</artifactId><version>1.13.2</version></dependency>
复制代码

创建 Kafka 源表的 SQL 语句如下:

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  -- 定义ts列上的水印, 将ts标记为事件时间属性) WITH (    'connector' = 'kafka', -- 使用kafka连接器    'topic' = 'user_behavior', -- kafka topic    'scan.startup.mode' = 'earliest-offset', -- 从头开始读取    'properties.Bootstrap.servers' = 'kafka:9094', -- kafka broker 地址    'format' = 'json'        -- 数据格式);
复制代码

在上面的 SQL 语句中,按照数据的格式声明了 5 个字段,除此之外,还通过计算列语法和 PROCTIME() 内置函数声明了一个产生处理时间的虚拟列。另外通过 WATERMARK 语法,在 ts 字段上声明了 watermark 策略(容忍 5s 乱序),ts 字段因此也成了事件时间列。

接下来,编写流处理代码。

Scala 代码实现:

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.table.api._import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
object UserBehaviorDemo {
def main(args: Array[String]) {//设置流执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment
//创建表环境 val tEnv = StreamTableEnvironment.create(env)
//启用检查点//启用检查点,并设置并行度为1 env.enableCheckpointing(5000).setParallelism(1)
//创建Kafka源表 tEnv.executeSql( """ |CREATE TABLE user_behavior ( | user_id bigint, | item_id bigint, | category_id bigint, | behavior string, | behavior_time bigint, | ts AS TO_TIMESTAMP(FROM_UNIXTIME(behavior_time, 'yyyy-MM-dd HH:mm:ss')), | proctime AS PROCTIME(), | WATERMARK FOR ts AS ts - INTERVAL '5' SECONDS |) |with( | 'connector'='kafka', | 'topic'='user_behavior', | 'properties.Bootstrap.servers'='localhost:9092', | 'properties.group.id'='testGroup', | 'scan.startup.mode'='latest-offset', | 'format'='csv', | 'csv.ignore-parse-errors' = 'true', | 'csv.field-delimiter'=',' |) """.stripMargin)
//创建mysql sink表 tEnv.executeSql( """ |CREATE TABLE buy_cnt_per_hour ( | hour_of_day TIMESTAMP(3), | buy_cnt BIGINT |) |with( | 'connector'='JDBC', | 'url'='JDBC:mysql://localhost:3306/xueai8?useSSL=false', | 'table-name'='buy_cnt_per_hour', | 'driver'='com.mysql.JDBC.Driver', | 'username'='root', | 'password'='admin' |) """.stripMargin)
tEnv//读取源表 .from("user_behavior")//定义大小为5s,滑动为2s 的滑动窗口 .window(Tumble over 1.minute on $"ts" as $"w")//分组 .groupBy($"w")//聚合 .select($"w".start.as("hour_of_day"),$"item_id".count.as("buy_cnt"))//.execute.print()//写入sink表 .executeInsert("buy_cnt_per_hour") }}
复制代码

Java 代码如下:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.Tumble;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import org.apache.flink.table.expressions.TimeIntervalUnit;import org.apache.flink.util.TimeUtils;
import static org.apache.flink.table.api.Expressions.$;import static org.apache.flink.table.api.Expressions.lit;
public class UserBehaviorDemo {
public static void main(String[] args) throws Exception {//设置流执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//创建表环境 StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//启用检查点,并设置并行度为1 env.enableCheckpointing(1000).setParallelism(1);
//创建Kafka源表 tEnv.executeSql( "CREATE TABLE user_behavior (" + " user_id bigint," + " item_id bigint," + " category_id bigint," + " behavior string," + " behavior_time bigint," + " ts AS TO_TIMESTAMP(FROM_UNIXTIME(behavior_time, 'yyyy-MM-dd HH:mm:ss'))," + " proctime AS PROCTIME()," + " WATERMARK FOR ts AS ts - INTERVAL '5' SECOND" + ")" + "WITH (" + " 'connector'='kafka'," + " 'topic'='user_behavior'," + " 'properties.Bootstrap.servers'='192.168.190.133:9092'," + " 'properties.group.id'='testGroup'," + " 'scan.startup.mode'='latest-offset'," + " 'format'='csv'," + " 'csv.ignore-parse-errors' = 'true'," + " 'csv.field-delimiter'=','" + ")" );
//创建mysql sink表 tEnv.executeSql("CREATE TABLE buy_cnt_per_hour (" + " hour_of_day TIMESTAMP(3)," + " buy_cnt BIGINT" + ") WITH (" + " 'connector' = 'JDBC'," + " 'url' = 'JDBC:mysql://localhost:3306/xueai8?useSSL=false'," + " 'table-name' = 'buy_cnt_per_hour'," + " 'driver' = 'com.mysql.JDBC.Driver'," + " 'username' = 'root'," + " 'password' = 'admin'" + ")" );
//统计每分钟的成交量 /*统计每小时的成交量就是每小时共有多少 "buy" 的用户行为。因此会需要用到 TUMBLE 窗口函数,按照一小时切窗。,然后每个窗口分别统计 "buy" 的个数,这可以通过先过滤出 "buy" 的数据,然后 COUNT(*) 实现。 SELECT HOUR(TUMBLE_START(ts, INTERVAL '1' HOUR)), COUNT(*) FROM user_behavior WHERE behavior = 'buy' GROUP BY TUMBLE(ts, INTERVAL '1' HOUR); */ /* 在MySQL数据库中创建相应的结果表: CREATE TABLE xueai8.buy_cnt_per_hour ( hour_of_day timestamp, buy_cnt bigint ) */
tEnv//读取源表 .from("user_behavior")//过滤出购买行为 .filter($("behavior").isEqual("buy"))//定义滚动窗口 .window(Tumble.over(lit(1).minute()).on($("ts")).as("w"))//分组 .groupBy($("w"))//聚合 .select($("w").start().as("hour_of_day"),$("item_id").count().as("buy_cnt"))//使用.extract(TimeIntervalUnit.MINUTE)抽取日期中的指定部分 .select($("w").start().extract(TimeIntervalUnit.MINUTE).as("hour_of_day"),$("item_id").count().as("buy_cnt")) .execute().collect().forEachRemaining(System.out::println);//写入sink表 .executeInsert("buy_cnt_per_hour"); }
}
复制代码

04、执行过程

请按以下步骤执行程序。

(1) 启动 Kafka。

首先打开一个终端,运行 ZooKeeper,命令如下:

$ ./bin/zookeeper-server-start.sh ./config/zookeeper.properties
复制代码


另打开一个终端,运行 Kafka 服务器,命令如下:

$ ./bin/kafka-server-start.sh ./config/server.properties
复制代码


再打开一个终端,创建名为 user_behavior 的 Kafka 主题,命令如下:

$./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic user_behavior
复制代码


查看已经存在的主题,命令如下:

$ ./bin/kafka-topics.sh --list --zookeeper localhost:2181
复制代码


(2) 在 MySQL 上创建接收表,SQL 语句如下:

CREATE TABLE buy_cnt_per_hour (  hour_of_day timestamp,  buy_cnt bigint);
复制代码


(3) 运行可视化终端 Grafana。

Grafana 是一款用 Go 语言开发的开源数据可视化工具,可以做数据监控和数据统计,带有告警功能。

首先安装 Grafana,安装步骤如下:

① 下载并安装包,然后解压缩即可。

② 启动。在命令行启动 Grafana 服务器,命令如下:

E:\BigData\Grafana\grafana-7.5.0\bin>grafana-server.exe
复制代码


③ 然后打开浏览器,访问地址 http://localhost:3000/。

④ 在 Grafana 中要先配置好数据源,指向 MySQL 中的接收表,如图 1 所示。


■图 1 在 Grafana 中要先配置好数据源

⑤ 在 Grafana 中创建 dashboard,查询获得数据,使用的 SQL 语句如下:

SELECTUNIX_TIMESTAMP(hour_of_day) as time_sec,     buy_cnt as valueFROM xueai8.buy_cnt_per_hour
复制代码


为了更好地看到动态变化的效果,设置一个仪表板显示的时间范围,如图 2 所示。


■ 图 2 在 Grafana 中设置一个仪表板显示的时间范围

(4) 运行 Flink 流程序。

(5) 执行数据生产者脚本 streamuserbehavior.sh。它调用 Kafka 自带的生产者脚本,以每秒 10 条的速度将数据发送给 Kafka 的 user_behavior 主题。编辑生产数据的脚本文件 streamuserbehavior.sh,代码如下:

#!/bin/bashBROKER=$1if [ -z "$1" ]; then        BROKER="localhost:9092"fi
cat UserBehavior_part.csv | while read line; do echo $line sleep 0.1done | ~/bigdata/kafka_2.11-2.4.1/bin/kafka-console-producer.sh --broker-list $BROKER --topic user_behavior
复制代码

执行该脚本,使用的命令如下

$ ./streamuserbehavior.sh
复制代码


(6) 观察 Grafana 中数据实时呈现效果,如图 3 所示。


■ 图 3 用户购买量实时统计结果显示

发布于: 刚刚阅读数: 4
用户头像

TiAmo

关注

有能力爱自己,有余力爱别人! 2022-06-16 加入

CSDN全栈领域优质创作者,万粉博主;阿里云专家博主、星级博主、技术博主、阿里云问答官,阿里云MVP;华为云享专家;华为Iot专家;

评论

发布
暂无评论
Flink实例:电商用户行为实时分析_数据分析_TiAmo_InfoQ写作社区