数据准备
这里我们准备了一份淘宝用户行为数据集(来自阿里云天池公开数据集)。本数据集包含了淘宝上某一天随机一百万用户的所有行为(包括点击、购买、加购、收藏)。数据集的组织形式和MovieLens-20M类似,即数据集的每一行表示一条用户行为,由用户ID、商品ID、商品类目ID、行为类型和时间戳组成,并以逗号分隔。关于数据集中每一列的详细描述如下:
列名称 | 说明 |
---|---|
用户ID | 整数类型,加密后的用户ID |
商品ID | 整数类型,加密后的商品ID |
商品类目ID | 整数类型,加密后的商品所属类目ID |
行为类型 | 字符串,枚举类型,包括(‘pv’, ‘buy’, ‘cart’, ‘fav’) |
时间戳 | 行为发生的时间戳,单位秒 |
你可以通过下面的命令下载数据集到项目的 resources
目录下:
$ cd my-flink-project/src/main/resources
$ curl https://raw.githubusercontent.com/wuchong/my-flink-project/master/src/main/resources/UserBehavior.csv > UserBehavior.csv
这里是否使用 curl 命令下载数据并不重要,你也可以使用 wget 命令或者直接访问链接下载数据。关键是,将数据文件保存到项目的 resources
目录下,方便应用程序访问。
编写程序
创建模拟数据源
我们先创建一个 UserBehavior
的 POJO 类(所有成员变量声明成public
便是POJO类),强类型化后能方便后续的处理。
/**
* 用户行为数据结构
**/
public static class UserBehavior {
public long userId; // 用户ID
public long itemId; // 商品ID
public int categoryId; // 商品类目ID
public String behavior; // 用户行为, 包括("pv", "buy", "cart", "fav")
public long timestamp; // 行为发生的时间戳,单位秒
}
接下来我们就可以创建一个 PojoCsvInputFormat
了, 这是一个读取 csv 文件并将每一行转成指定 POJO
类型(在我们案例中是 UserBehavior
)的输入器。
// UserBehavior.csv 的本地文件路径
URL fileUrl = HotItems2.class.getClassLoader().getResource("UserBehavior.csv");
Path filePath = Path.fromLocalFile(new File(fileUrl.toURI()));
// 抽取 UserBehavior 的 TypeInformation,是一个 PojoTypeInfo
PojoTypeInfo<UserBehavior> pojoType = (PojoTypeInfo<UserBehavior>) TypeExtractor.createTypeInfo(UserBehavior.class);
// 由于 Java 反射抽取出的字段顺序是不确定的,需要显式指定下文件中字段的顺序
String[] fieldOrder = new String[]{"userId", "itemId", "categoryId", "behavior", "timestamp"};
// 创建 PojoCsvInputFormat
PojoCsvInputFormat<UserBehavior> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);
下一步我们用 PojoCsvInputFormat
创建输入源。
DataStream<UserBehavior> dataSource = env.createInput(csvInput, pojoType);
这就创建了一个 UserBehavior
类型的 DataStream
。
EventTime与Watermark
当我们说“统计过去一小时内点击量”,这里的“一小时”是指什么呢? 在 Flink 中它可以是指 ProcessingTime ,也可以是 EventTime,由用户决定。
- ProcessingTime:事件被处理的时间。也就是由机器的系统时间来决定
- EventTime:事件发生的时间。一般就是数据本身携带的时间
在本案例中,我们需要统计业务时间上的每小时的点击量,所以要基于 EventTime 来处理。那么如果让 Flink 按照我们想要的业务时间来处理呢?这里主要有两件事情要做:
告诉 Flink 我们现在按照 EventTime 模式进行处理,Flink 默认使用 ProcessingTime 处理,所以我们要显式设置下。
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
- 指定如何获得业务时间,以及生成 Watermark。Watermark 是用来追踪业务事件的概念,可以理解成 EventTime 世界中的时钟,用来指示当前处理到什么时刻的数据了。由于我们的数据源的数据已经经过整理,没有乱序,即事件的时间戳是单调递增的,所以可以将每条数据的业务时间就当做 Watermark。这里我们用
AscendingTimestampExtractor
来实现时间戳的抽取和 Watermark 的生成。
注意:真实业务场景一般都是存在乱序的,所以一般使用 BoundedOutOfOrdernessTimestampExtractor
。
DataStream<UserBehavior> timedData = dataSource
.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
@Override
public long extractAscendingTimestamp(UserBehavior userBehavior) {
// 原始数据单位秒,将其转成毫秒
return userBehavior.timestamp * 1000;
}
});
这样我们就得到了一个带有时间标记的数据流了,后面就能做一些窗口的操作。
过滤出点击事件
在开始窗口操作之前,先回顾下需求“每隔5分钟输出过去一小时内点击量最多的前 N 个商品”。由于原始数据中存在点击、加购、购买、收藏各种行为的数据,但是我们只需要统计点击量,所以先使用 FilterFunction
将点击行为数据过滤出来。
DataStream<UserBehavior> pvData = timedData
.filter(new FilterFunction<UserBehavior>() {
@Override
public boolean filter(UserBehavior userBehavior) throws Exception {
// 过滤出只有点击的数据
return userBehavior.behavior.equals("pv");
}
});
窗口统计点击量
由于要每隔5分钟统计一次最近一小时每个商品的点击量,所以窗口大小是一小时,每隔5分钟滑动一次。即分别要统计 [09:00, 10:00), [09:05, 10:05), [09:10, 10:10)… 等窗口的商品点击量。是一个常见的滑动窗口需求(Sliding Window)。
DataStream<ItemViewCount> windowedData = pvData
// 对商品进行分组
.keyBy("itemId")
// 对每个商品做滑动窗口(1小时窗口,5分钟滑动一次)
.timeWindow(Time.minutes(60), Time.minutes(5))
// 做增量的聚合操作,它能使用AggregateFunction提前聚合掉数据,减少 state 的存储压力
.aggregate(new CountAgg(), new WindowResultFunction());
CountAgg
这里的CountAgg
实现了AggregateFunction
接口,功能是统计窗口中的条数,即遇到一条数据就加一。
/**
* COUNT 统计的聚合函数实现,每出现一条记录加一
**/
public static class CountAgg implements AggregateFunction<UserBehavior, Long, Long> {
@Override
public Long createAccumulator() {
return 0L;
}
@Override
public Long add(UserBehavior userBehavior, Long acc) {
return acc + 1;
}
@Override
public Long getResult(Long acc) {
return acc;
}
@Override
public Long merge(Long acc1, Long acc2) {
return acc1 + acc2;
}
}
WindowFunction
.aggregate(AggregateFunction af, WindowFunction wf)
的第二个参数WindowFunction
将每个 key每个窗口聚合后的结果带上其他信息进行输出。这里实现的WindowResultFunction
将主键商品ID,窗口,点击量封装成了ItemViewCount
进行输出。
/**
* 用于输出窗口的结果
**/
public static class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {
@Override
public void apply(
Tuple key, // 窗口的主键,即 itemId
TimeWindow window, // 窗口
Iterable<Long> aggregateResult, // 聚合函数的结果,即 count 值
Collector<ItemViewCount> collector // 输出类型为 ItemViewCount
) throws Exception {
Long itemId = ((Tuple1<Long>) key).f0;
Long count = aggregateResult.iterator().next();
collector.collect(ItemViewCount.of(itemId, window.getEnd(), count));
}
}
/**
* 商品点击量(窗口操作的输出类型)
**/
public static class ItemViewCount {
public long itemId; // 商品ID
public long windowEnd; // 窗口结束时间戳
public long viewCount; // 商品的点击量
public static ItemViewCount of(long itemId, long windowEnd, long viewCount) {
ItemViewCount result = new ItemViewCount();
result.itemId = itemId;
result.windowEnd = windowEnd;
result.viewCount = viewCount;
return result;
}
}
现在我们得到了每个商品在每个窗口的点击量的数据流。