Flink 零基礎實戰教程：如何計算實時熱門商品

在上一篇入門教程中，我們已經能夠快速構建一個基礎的 Flink 程序了。本文會一步步地帶領你實現一個更複雜的 Flink 應用程序：實時熱門商品。在開始本文前我們建議你先實踐一遍上篇文章，因為本文會沿用上文的my-flink-project項目框架。

通過本文你將學到：

如何基於 EventTime 處理，如何指定 Watermark
如何使用 Flink 靈活的 Window API
何時需要用到 State，以及如何使用
如何使用 ProcessFunction 實現 TopN 功能

實戰案例介紹

「實時熱門商品」的需求，我們可以將「實時熱門商品」翻譯成程序員更好理解的需求：每隔5分鐘輸出最近一小時內點擊量最多的前 N 個商品。將這個需求進行分解我們大概要做這麼幾件事情：

抽取出業務時間戳，告訴 Flink 框架基於業務時間做窗口
過濾出點擊行為數據
按一小時的窗口大小，每5分鐘統計一次，做滑動窗口聚合（Sliding Window）
按每個窗口聚合，輸出每個窗口中點擊量前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<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") .timeWindow(Time.minutes(60), Time.minutes(5)) .aggregate(new CountAgg(), new WindowResultFunction());

我們使用.keyBy("itemId")對商品進行分組，使用.timeWindow(Time size, Time slide)對每個商品做滑動窗口（1小時窗口，5分鐘滑動一次）。然後我們使用 .aggregate(AggregateFunction af, WindowFunction wf) 做增量的聚合操作，它能使用AggregateFunction提前聚合掉數據，減少 state 的存儲壓力。較之.apply(WindowFunction wf)會將窗口中的數據都存儲下來，最後一起計算要高效地多。aggregate()方法的第一個參數用於

這裡的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;
}
}

.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;
}
}

現在我們得到了每個商品在每個窗口的點擊量的數據流。

TopN 計算最熱門商品

為了統計每個窗口下最熱門的商品，我們需要再次按窗口進行分組，這裡根據ItemViewCount中的windowEnd進行keyBy()操作。然後使用 ProcessFunction 實現一個自定義的 TopN 函數 TopNHotItems 來計算點擊量排名前3名的商品，並將排名結果格式化成字元串，便於後續輸出。

DataStream<String> topItems = windowedData .keyBy("windowEnd") .process(new TopNHotItems(3)); // 求點擊量前3名的商品

ProcessFunction 是 Flink 提供的一個 low-level API，用於實現更高級的功能。它主要提供了定時器 timer 的功能（支持EventTime或ProcessingTime）。本案例中我們將利用 timer 來判斷何時收齊了某個 window 下所有商品的點擊量數據。由於 Watermark 的進度是全局的，

在 processElement 方法中，每當收到一條數據（ItemViewCount），我們就註冊一個 windowEnd+1 的定時器（Flink 框架會自動忽略同一時間的重複註冊）。windowEnd+1 的定時器被觸發時，意味著收到了windowEnd+1的 Watermark，即收齊了該windowEnd下的所有商品窗口統計值。我們在 onTimer() 中處理將收集的所有商品及點擊量進行排序，選出 TopN，並將排名信息格式化成字元串後進行輸出。

這裡我們還使用了 ListState 來存儲收到的每條 ItemViewCount 消息，保證在發生故障時，狀態數據的不丟失和一致性。ListState 是 Flink 提供的類似 Java List 介面的 State API，它集成了框架的 checkpoint 機制，自動做到了 exactly-once 的語義保證。

/** 求某個窗口中前 N 名的熱門點擊商品，key 為窗口時間戳，輸出為 TopN 的結果字元串 */ public static class TopNHotItems extends KeyedProcessFunction<Tuple, ItemViewCount, String> {

private final int topSize;

public TopNHotItems(int topSize) {
this.topSize = topSize;
}

// 用於存儲商品與點擊數的狀態，待收齊同一個窗口的數據後，再觸發 TopN 計算
private ListState<ItemViewCount> itemState;

@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// 狀態的註冊
ListStateDescriptor<ItemViewCount> itemsStateDesc = new ListStateDescriptor<>(
"itemState-state",
ItemViewCount.class);
itemState = getRuntimeContext().getListState(itemsStateDesc);
}

@Override
public void processElement(
ItemViewCount input,
Context context,
Collector<String> 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<String> out) throws Exception {
// 獲取收到的所有商品點擊量
List<ItemViewCount> allItems = new ArrayList<>();
for (ItemViewCount item : itemState.get()) {
allItems.add(item);
}
// 提前清除狀態中的數據，釋放空間
itemState.clear();
// 按照點擊量從大到小排序
allItems.sort(new Comparator<ItemViewCount>() {
@Override
public int compare(ItemViewCount o1, ItemViewCount o2) {
return (int) (o2.viewCount - o1.viewCount);
}
});
// 將排名信息格式化成 String, 便於列印
StringBuilder result = new StringBuilder();
result.append("====================================
");
result.append("時間: ").append(new Timestamp(timestamp-1)).append("
");
for (int i=0;i<topSize;i++) {
ItemViewCount currentItem = allItems.get(i);
// No1: 商品ID=12224 瀏覽量=2413
result.append("No").append(i).append(":")
.append(" 商品ID=").append(currentItem.itemId)
.append(" 瀏覽量=").append(currentItem.viewCount)
.append("
");
}
result.append("====================================