在上一篇入门教程中，我们早已就能快速建立一个基础的Flink程序了。本文会一步步地率领你实现一个更复杂的Flink应用程序：实时热门商品。在开始本文前我们建议你先实践一遍下篇文章，由于本文会承袭上文的my-flink-project项目框架。
通过本文你将学到：
实战案例介绍
“实时热门商品”的需求，我们可以将“实时热门商品”翻译成程序员更好理解的需求：每隔5分钟输出近来一小时内点击量最多的前N个商品。将这个需求进行分解我们大约要做如此几件事情：
数据打算
这儿我们打算了一份天猫用户行为数据集（来自阿里云天池公开数据集，非常谢谢）。本数据集包含了天猫上某三天随机一百万用户的所有行为（包括点击、购买、加购、收藏）。数据集的组织方式和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目录下，便捷应用程序访问。
编撰程序
在src/main/java/myflink下创建HotItems.java文件：
package myflink;
public class HotItems {
public static void main(String[] args) throws Exception {
}
}与上文一样，我们会一步步往上面填充代码。第一步一直是创建一个StreamExecutionEnvironment，我们把它添加到main函数中。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 为了打印到控制台的结果不乱序，我们配置全局的并发为1，这里改变并发对结果正确性没有影响
env.setParallelism(1);创建模拟数据源
在数据打算章节，我们早已将测试的数据集下载到本地了。因为是一个csv文件，我们将使用CsvInputFormat创建模拟数据源。
注：尽管一个流式应用应当是一个仍然运行着的程序，须要消费一个无限数据源。并且在本案例教程中，为了省去建立真实数据源的冗长，我们使用了文件来模拟真实数据源，这并不影响下文要介绍的知识点。这也是一种本地验证Flink应用程序正确性的常用方法。
我们先创建一个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 pojoType = (PojoTypeInfo) TypeExtractor.createTypeInfo(UserBehavior.class);
// 由于 Java 反射抽取出的字段顺序是不确定的，需要显式指定下文件中字段的顺序
String[] fieldOrder = new String[]{"userId", "itemId", "categoryId", "behavior", "timestamp"};
// 创建 PojoCsvInputFormat
PojoCsvInputFormat csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder); 下一步我们用PojoCsvInputFormat创建输入源。
DataStream dataSource = env.createInput(csvInput, pojoType); 这就创建了一个UserBehavior类型的DataStream。
EventTime与Watermark
当我们说“统计过去一小时内点击量”，这儿的“一小时”是指哪些呢？在Flink中它可以是指ProcessingTime，也可以是EventTime，由用户决定。
在本案例中，我们须要统计业务时间上的每小时的点击量，所以要基于EventTime来处理。这么假如让Flink根据我们想要的业务时间来处理呢？这儿主要有两件事情要做。
第一件是告诉Flink我们如今根据EventTime模式进行处理，Flink默认使用ProcessingTime处理，所以我们要显式设置下。
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);第二件事情是指定怎样获得业务时间，以及生成Watermark。Watermark是拿来追踪业务风波的概念，可以理解成EventTime世界中的时钟，拿来指示当前处理到哪些时刻的数据了。因为我们的数据源的数据早已经过整理，没有正序，即风波的时间戳是单调递增的，所以可以将每条数据的业务时间就当作Watermark。这儿我们用AscendingTimestampExtractor来实现时间戳的抽取和Watermark的生成。
注：真实业务场景通常都是存在正序的，所以通常使用BoundedOutOfOrdernessTimestampExtractor。
DataStream timedData = dataSource
.assignTimestampsAndWatermarks(new AscendingTimestampExtractor() {
@Override
public long extractAscendingTimestamp(UserBehavior userBehavior) {
// 原始数据单位秒，将其转成毫秒
return userBehavior.timestamp * 1000;
}
}); 这样我们就得到了一个带有时间标记的数据流了，前面才能做一些窗口的操作。
过滤出点击风波
在开始窗口操作之前，先回顾下需求“每隔5分钟输出过去一小时内点击量最多的前N个商品”。因为原始数据中存在点击、加购、购买、收藏各类行为的数据，并且我们只须要统计点击量，所以先使用FilterFunction将点击行为数据过滤下来。
DataStream pvData = timedData
.filter(new FilterFunction() {
@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)…等窗口的商品点击量。是一个常见的滑动窗口需求（SlidingWindow）。
DataStream windowedData = pvData
.keyBy("itemId")
.timeWindow(Time.minutes(60), Time.minutes(5))
.aggregate(new CountAgg(), new WindowResultFunction()); 我们使用.keyBy("itemId")对商品进行分组，使用.timeWindow(Timesize,Timeslide)对每位商品做滑动窗口（1小时窗口，5分钟滑动一次）。之后我们使用.aggregate(AggregateFunctionaf,WindowFunctionwf)做增量的聚合操作，它能使用AggregateFunction提早聚合掉数据，降低state的储存压力。较之.apply(WindowFunctionwf)会将窗口中的数据都储存出来，最后一起估算要高效地多。aggregate()方式的第一个参数用于
这儿的CountAgg实现了AggregateFunction插口，功能是统计窗口中的条数，即遇见一条数据就加一。
/** COUNT 统计的聚合函数实现，每出现一条记录加一 */
public static class CountAgg implements AggregateFunction {
@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;
}
} .aggregate(AggregateFunctionaf,WindowFunctionwf)的第二个参数WindowFunction将每位key每位窗口聚合后的结果带上其他信息进行输出。我们这儿实现的WindowResultFunction将字段商品ID，窗口，点击量封装成了ItemViewCount进行输出。
/** 用于输出窗口的结果 */
public static class WindowResultFunction implements WindowFunction {
@Override
public void apply(
Tuple key, // 窗口的主键，即 itemId
TimeWindow window, // 窗口
Iterable aggregateResult, // 聚合函数的结果，即 count 值
Collector collector // 输出类型为 ItemViewCount
) throws Exception {
Long itemId = ((Tuple1) 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;
}
} 如今我们得到了每位商品在每位窗口的点击量的数据流。
TopN估算最热门商品
为了统计每位窗口下最热门的商品，我们须要再度按窗口进行分组，这儿按照ItemViewCount中的windowEnd进行keyBy()操作。之后使用ProcessFunction实现一个自定义的TopN函数TopNHotItems来估算点击量排行前3名的商品，并将排行结果格式化成字符串，以便后续输出。
DataStream topItems = windowedData
.keyBy("windowEnd")
.process(new TopNHotItems(3)); // 求点击量前3名的商品 ProcessFunction是Flink提供的一个low-levelAPI，用于实现更中级的功能。它主要提供了定时器timer的功能（支持EventTime或ProcessingTime）。本案例中我们将借助timer来判定何时收齐了某个window下所有商品的点击量数据。因为Watermark的进度是全局的，
在processElement方式中，每每收到一条数据（ItemViewCount），我们就注册一个windowEnd+1的定时器（Flink框架会手动忽视同一时间的重复注册）。windowEnd+1的定时器被触发时，意味着收到了windowEnd+1的Watermark，即收齐了该windowEnd下的所有商品窗口统计值。我们在onTimer()中处理将搜集的所有商品及点击量进行排序，选出TopN，并将排行信息格式化成字符串后进行输出。
这儿我们还使用了ListState来储存收到的每条ItemViewCount消息，保证在发生故障时，状态数据的不遗失和一致性。ListState是Flink提供的类似JavaList插口的StateAPI，它集成了框架的checkpoint机制，手动做到了exactly-once的语义保证。
/** 求某个窗口中前 N 名的热门点击商品，key 为窗口时间戳，输出为 TopN 的结果字符串 */
public static class TopNHotItems extends KeyedProcessFunction {
private final int topSize;
public TopNHotItems(int topSize) {
this.topSize = topSize;
}
// 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算
private ListState itemState;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// 状态的注册
ListStateDescriptor itemsStateDesc = new ListStateDescriptor<>(
"itemState-state",
ItemViewCount.class);
itemState = getRuntimeContext().getListState(itemsStateDesc);
}
@Override
public void processElement(
ItemViewCount input,
Context context,
Collector collector) throws Exception {
// 每条数据都保存到状态中
itemState.add(input);
// 注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收齐了属于windowEnd窗口的所有商品数据
context.timerService().registerEventTimeTimer(input.windowEnd + 1);
}
@Override
public void onTimer(
long timestamp, OnTimerContext ctx, Collector out) throws Exception {
// 获取收到的所有商品点击量
List allItems = new ArrayList<>();
for (ItemViewCount item : itemState.get()) {
allItems.add(item);
}
// 提前清除状态中的数据，释放空间
itemState.clear();
// 按照点击量从大到小排序
allItems.sort(new Comparator() {
@Override
public int compare(ItemViewCount o1, ItemViewCount o2) {
return (int) (o2.viewCount - o1.viewCount);
}
});
// 将排名信息格式化成 String, 便于打印
StringBuilder result = new StringBuilder();
result.append("====================================\n");
result.append("时间: ").append(new Timestamp(timestamp-1)).append("\n");
for (int i=0;i 复印输出
最后一步我们将结果复印输出到控制台，并调用env.execute执行任务。
topItems.print();
env.execute("Hot Items Job");运行程序
直接运行main函数，能够看见不断输出的每位时间点的热门商品ID。
总结
本文的完整代码可以通过GitHub访问到。本文通过实现一个“实时热门商品”的案例，学习和实践了Flink的多个核心概念和API用法。包括EventTime、Watermark的使用，State的使用，WindowAPI的使用，以及TopN的实现。希望本文能加深你们对Flink的理解，帮助你们解决实战上遇见的问题。
整代码请移步GitHub访问：
