Flink基礎：實時處理管道與ETL
往期推薦：
Flink基礎：入門介紹
Flink基礎：DataStream API
Flink深入淺出：資源管理
Flink深入淺出：部署模式
Flink深入淺出：記憶體模型
Flink深入淺出：JDBC Source從理論到實戰
Flink深入淺出：Sql Gateway原始碼分析
Flink深入淺出：JDBC Connector原始碼分析
Flink的經典使用場景是ETL，即Extract抽取、Transform轉換、Load載入，可以從一個或多個數據源讀取資料，經過處理轉換後，儲存到另一個地方，本篇將會介紹如何使用DataStream API來實現這種應用。注意Flink Table和SQL
1 無狀態的轉換
無狀態即不需要在操作中維護某個中間狀態，典型的例子如map和flatmap。
map()
下面是一個轉換操作的例子，需要根據輸入資料建立一個計程車起始位置和目標位置的物件。首先定義出租車的位置物件：
public static class EnrichedRide extends TaxiRide {
public int startCell;
public int endCell;
public EnrichedRide() {}
public EnrichedRide(TaxiRide ride) {
this.rideId = ride.rideId;
this.isStart = ride.isStart;
...
this.startCell = GeoUtils.mapToGridCell(ride.startLon, ride.startLat);
this.endCell = GeoUtils.mapToGridCell(ride.endLon, ride.endLat);
}
public String toString() {
return super.toString() + "," +
Integer.toString(this.startCell) + "," +
Integer.toString(this.endCell);
}
} 使用的時候可以註冊一個MapFunction，該函式接收TaxiRide物件，輸出EnrichRide物件。
public static class Enrichment implements MapFunction<TaxiRide, EnrichedRide> {
@Override
public EnrichedRide map(TaxiRide taxiRide) throws Exception {
return new EnrichedRide(taxiRide);
}
}
使用時只需要建立map物件即可：
DataStream<TaxiRide> rides = env.addSource(new TaxiRideSource(...));
DataStream<EnrichedRide> enrichedNYCRides = rides
.filter(new RideCleansingSolution.NYCFilter())
.map(new Enrichment());
enrichedNYCRides.print();
flatmap()
MapFunction適合一對一的轉換，對於輸入流的每個元素都有一個元素輸出。如果需要一對多的場景，可以使用flatmap：
DataStream<TaxiRide> rides = env.addSource(new TaxiRideSource(...));
DataStream<EnrichedRide> enrichedNYCRides = rides
.flatMap(new NYCEnrichment());
enrichedNYCRides.print();
FlatMapFunction的定義：
public static class NYCEnrichment implements FlatMapFunction<TaxiRide, EnrichedRide> {
@Override
public void flatMap(TaxiRide taxiRide, Collector<EnrichedRide> out) throws Exception {
FilterFunction<TaxiRide> valid = new RideCleansing.NYCFilter();
if (valid.filter(taxiRide)) {
out.collect(new EnrichedRide(taxiRide));
}
}
}
通過collector，可以在flatmap中任意新增零個或多個元素。
2 Keyed Streams
keyBy()
有時需要對資料流按照某個欄位進行分組，每個事件會根據該欄位相同的值彙總到一起。比如，希望查詢相同出發位置的路線。如果在SQL中可能會使用GROUP BY startCell，在Flink中可以直接使用keyBy函式：
rides
.flatMap(new NYCEnrichment())
.keyBy(value -> value.startCell)
keyBy會引起重分割槽而導致網路資料shuffle，通常這種代價都很昂貴，因為每次shuffle時需要進行資料的序列化和反序列化，既浪費CPU資源，又佔用網路頻寬。
通過對startCell進行分組，這種方式的分組可能會由於編譯器而丟失欄位的型別資訊，因此Flink也支援把欄位包裝成Tuple，基於元素位置進行分組。當然也支援使用KeySelector函式，自定義分組規則。
rides
.flatMap(new NYCEnrichment())
.keyBy(
new KeySelector<EnrichedRide, int>() {
@Override
public int getKey(EnrichedRide enrichedRide) throws Exception {
return enrichedRide.startCell;
}
})
可以直接使用lambda表示式：
rides
.flatMap(new NYCEnrichment())
.keyBy(enrichedRide -> enrichedRide.startCell)
key可以自定義計算規則
keyselector不限制從必須從事件中抽取key，也可以自定義任何計算key的方法。但需要保證輸出的key是一致的，並且實現了對應的hashCode和equals方法。生成key的規則一定要穩定，因為生成key可能在應用執行的任何時間，因此一定要保證key生成規則的持續穩定。
key可以通過某個欄位選擇：
keyBy(enrichedRide -> enrichedRide.startCell)
也可以直接替換成某個方法：
keyBy(ride -> GeoUtils.mapToGridCell(ride.startLon, ride.startLat))
Keyed Stream的聚合
下面的例子中，建立了一個包含startCell和花費時間的二元組：
import org.joda.time.Interval;
DataStream<Tuple2<Integer, Minutes>> minutesByStartCell = enrichedNYCRides
.flatMap(new FlatMapFunction<EnrichedRide, Tuple2<Integer, Minutes>>() {
@Override
public void flatMap(EnrichedRide ride,
Collector<Tuple2<Integer, Minutes>> out) throws Exception {
if (!ride.isStart) {
Interval rideInterval = new Interval(ride.startTime, ride.endTime);
Minutes duration = rideInterval.toDuration().toStandardMinutes();
out.collect(new Tuple2<>(ride.startCell, duration));
}
}
});
現在需要輸出每個起始位置最長距離的路線，有很多種方式可以實現。以上面的資料為例，可以通過startcell進行聚合，然後選擇時間最大的元素輸出：
minutesByStartCell .keyBy(value -> value.f0) // .keyBy(value -> value.startCell) .maxBy(1) // duration .print();
可以得到輸出結果：
4> (64549,5M) 4> (46298,18M) 1> (51549,14M) 1> (53043,13M) 1> (56031,22M) 1> (50797,6M) ... 1> (50797,8M) ... 1> (50797,11M) ... 1> (50797,12M)
狀態
上面是一個有狀態的例子，Flink需要記錄每個key的最大值。無論何時在應用中涉及到狀態，都需要考慮這個狀態有多大。如果key的空間是無限大的，那麼flink可能需要維護大量的狀態資訊。當使用流時，一定要對無限視窗的聚合十分敏感，因為它是對整個流進行操作，很有可能因為維護的狀態資訊不斷膨脹，而導致記憶體溢位。在上面使用的maxBy就是經典的的聚合操作，也可以使用更通用的reduce來自定義聚合方法。
3 有狀態的操作
Flink針對狀態的管理有很多易用的特性，比如：
- 支援本地儲存：基於程序記憶體來儲存狀態
- 狀態的持久化：定期儲存到檢查點，保證容錯
- 垂直擴充套件：Flink狀態可以把狀態儲存到RocksDB中，也支援擴充套件到本地磁碟
- 水平擴充套件：狀態支援在叢集中擴縮容，通過調整並行度，自動拆分狀態
- 可查詢：Flink的狀態可以在外部直接查詢
Rich函式
Flink有幾種函式介面，包括FilterFunction, MapFunction，FlatMapFunction等。對於每個介面，Flink都提供了對應的Rich方法。比如RichFlatMapFunction，提供了額外的一些方法：
- open(Configuration c) 在初始化的時候呼叫一次，用於載入靜態資料，開啟外部服務的連線等
- close() 流關閉時呼叫
- getRuntimeContext() 提供進入全域性狀態的方法，需要了解如何建立和查詢狀態
使用Keyed State的例子
下面是一個針對事件的key進行去重的例子：
private static class Event {
public final String key;
public final long timestamp;
...
}
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(new EventSource())
.keyBy(e -> e.key)
.flatMap(new Deduplicator())
.print();
env.execute();
}
為了實現這個功能，deduplicator需要記住一些資訊，對於每個key，都需要記錄是否已經存在。Flink支援幾種不同型別的狀態，最簡單的一種是valueState。對於每個key，flink都為它儲存一個物件，在上面的例子中物件是Boolean。Deduplicator有兩個方法：open()和flatMap()。open方法通過descriptor為狀態起了一個標識名稱，並宣告型別為Boolean。
public static class Deduplicator extends RichFlatMapFunction<Event, Event> {
ValueState<Boolean> keyHasBeenSeen;
@Override
public void open(Configuration conf) {
ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor<>("keyHasBeenSeen", Types.BOOLEAN);
keyHasBeenSeen = getRuntimeContext().getState(desc);
}
@Override
public void flatMap(Event event, Collector<Event> out) throws Exception {
if (keyHasBeenSeen.value() == null) {
out.collect(event);
keyHasBeenSeen.update(true);
}
}
}
flatMap中呼叫state.value()獲取狀態。flink在上下文中為每個key儲存了一個狀態值，只有當值為null時，說明這個key之前沒有出現過，然後將其更新為true。當flink呼叫open時，狀態是空的。但是當呼叫flatMap時，key可以通過context進行訪問。當在叢集模式中執行時，會有很多個Deduplicator例項，每個負責維護一部分key的事件。因此，當使用單個事件的valuestate時，要理解它背後其實不是一個值，而是每個key都對應一個狀態值，並且分散式的儲存在叢集中的各個節點程序上。
清除狀態
有時候key的空間可能是無限制的，flink會為每個key儲存一個boolean物件。如果key的數量是有限的還好，但是應用往往是持續不間斷的執行，那麼key可能會無限增長，因此需要清理不再使用的key。可以通過state.clear()進行清理。比如針對某個key按照某一時間頻率進行清理，在processFunction中可以瞭解到如何在事件驅動的應用中執行定時器操作。也可以在狀態描述符中為狀態設定TTL生存時間，這樣狀態可以自動進行清理。
非keyed狀態
狀態也支援在非key型別的上下文中使用，這種叫做操作符狀態，operator state。典型的場景是Flink讀取Kafka時記錄的offset資訊。
4 連線流
大部分場景中Flink都是接收一個數據流輸出一個數據流，類似管道式的處理資料：
也有的場景需要動態的修改函式中的資訊，比如閾值、規則或者其他的引數，這種設計叫做connected streams，流會擁有兩個輸入，類似：
在下面的例子中，通過控制流用來指定必須過濾的單詞：
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> control = env.fromElements("DROP", "IGNORE").keyBy(x -> x);
DataStream<String> streamOfWords = env.fromElements("Apache", "DROP", "Flink", "IGNORE").keyBy(x -> x);
control
.connect(datastreamOfWords)
.flatMap(new ControlFunction())
.print();
env.execute();
}
兩個流可以通過key的方式連線，keyby用來分組資料，這樣保證相同型別的資料可以進入到相同的例項中。上面的例子兩個流都是字串，
public static class ControlFunction extends RichCoFlatMapFunction<String, String, String> {
private ValueState<Boolean> blocked;
@Override
public void open(Configuration config) {
blocked = getRuntimeContext().getState(new ValueStateDescriptor<>("blocked", Boolean.class));
}
@Override
public void flatMap1(String control_value, Collector<String> out) throws Exception {
blocked.update(Boolean.TRUE);
}
@Override
public void flatMap2(String data_value, Collector<String> out) throws Exception {
if (blocked.value() == null) {
out.collect(data_value);
}
}
}
blocked用於記錄key的控制邏輯，key的state會在兩個流間共享。flatMap1和flatMap2會被兩個流呼叫，分別用來更新和獲取狀態，從而實現通過一個流控制另一個流的目的。
總結：本片從狀態上講述了有狀態的操作和無狀態的操作，還介紹了狀態的使用以及連線流的適用場景。後面會介紹DataStream的操作和狀態的管理。
&n