摘要：這些切片稱為窗口。函數允許對常規數據流進行分組。通常，這是非并行數據轉換，因為它在非分區數據流上運行。
前言
在第一篇介紹 Flink 的文章 《《從0到1學習Flink》—— Apache Flink 介紹》 中就說過 Flink 程序的結構
Flink 應用程序結構就是如上圖所示：
1、Source: 數據源，Flink 在流處理和批處理上的 source 大概有 4 類：基于本地集合的 source、基于文件的 source、基于網絡套接字的 source、自定義的 source。自定義的 source 常見的有 Apache kafka、Amazon Kinesis Streams、RabbitMQ、Twitter Streaming API、Apache NiFi 等，當然你也可以定義自己的 source。
2、Transformation：數據轉換的各種操作，有 Map / FlatMap / Filter / KeyBy / Reduce / Fold / Aggregations / Window / WindowAll / Union / Window join / Split / Select / Project 等，操作很多，可以將數據轉換計算成你想要的數據。
3、Sink：接收器，Flink 將轉換計算后的數據發送的地點 ，你可能需要存儲下來，Flink 常見的 Sink 大概有如下幾類：寫入文件、打印出來、寫入 socket 、自定義的 sink 。自定義的 sink 常見的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem 等，同理你也可以定義自己的 Sink。
在上四篇文章介紹了 Source 和 Sink：
1、《從0到1學習Flink》—— Data Source 介紹
2、《從0到1學習Flink》—— 如何自定義 Data Source ？
3、《從0到1學習Flink》—— Data Sink 介紹
4、《從0到1學習Flink》—— 如何自定義 Data Sink ？
那么這篇文章我們就來看下 Flink Data Transformation 吧，數據轉換操作還是蠻多的，需要好好講講！
Transformation
Map
這是最簡單的轉換之一，其中輸入是一個數據流，輸出的也是一個數據流：
還是拿上一篇文章的案例來將數據進行 map 轉換操作：
SingleOutputStreamOperator map = student.map(new MapFunction() {
@Override
public Student map(Student value) throws Exception {
Student s1 = new Student();
s1.id = value.id;
s1.name = value.name;
s1.password = value.password;
s1.age = value.age + 5;
return s1;
}
});
map.print();
將每個人的年齡都增加 5 歲，其他不變。
FlatMap
FlatMap 采用一條記錄并輸出零個，一個或多個記錄。
SingleOutputStreamOperator flatMap = student.flatMap(new FlatMapFunction() {
@Override
public void flatMap(Student value, Collector out) throws Exception {
if (value.id % 2 == 0) {
out.collect(value);
}
}
});
flatMap.print();
這里將 id 為偶數的聚集出來。
Filter
Filter 函數根據條件判斷出結果。
SingleOutputStreamOperator filter = student.filter(new FilterFunction() {
@Override
public boolean filter(Student value) throws Exception {
if (value.id > 95) {
return true;
}
return false;
}
});
filter.print();
這里將 id 大于 95 的過濾出來，然后打印出來。
KeyBy
KeyBy 在邏輯上是基于 key 對流進行分區。在內部，它使用 hash 函數對流進行分區。它返回 KeyedDataStream 數據流。
KeyedStream keyBy = student.keyBy(new KeySelector() {
@Override
public Integer getKey(Student value) throws Exception {
return value.age;
}
});
keyBy.print();
上面對 student 的 age 做 KeyBy 操作分區
Reduce
Reduce 返回單個的結果值，并且 reduce 操作每處理一個元素總是創建一個新值。常用的方法有 average, sum, min, max, count，使用 reduce 方法都可實現。
SingleOutputStreamOperator reduce = student.keyBy(new KeySelector() {
@Override
public Integer getKey(Student value) throws Exception {
return value.age;
}
}).reduce(new ReduceFunction() {
@Override
public Student reduce(Student value1, Student value2) throws Exception {
Student student1 = new Student();
student1.name = value1.name + value2.name;
student1.id = (value1.id + value2.id) / 2;
student1.password = value1.password + value2.password;
student1.age = (value1.age + value2.age) / 2;
return student1;
}
});
reduce.print();
上面先將數據流進行 keyby 操作，因為執行 reduce 操作只能是 KeyedStream，然后將 student 對象的 age 做了一個求平均值的操作。
Fold
Fold 通過將最后一個文件夾流與當前記錄組合來推出 KeyedStream。 它會發回數據流。
KeyedStream.fold("1", new FoldFunction() {
@Override
public String fold(String accumulator, Integer value) throws Exception {
return accumulator + "=" + value;
}
})
Aggregations
DataStream API 支持各種聚合，例如 min，max，sum 等。 這些函數可以應用于 KeyedStream 以獲得 Aggregations 聚合。
KeyedStream.sum(0)
KeyedStream.sum("key")
KeyedStream.min(0)
KeyedStream.min("key")
KeyedStream.max(0)
KeyedStream.max("key")
KeyedStream.minBy(0)
KeyedStream.minBy("key")
KeyedStream.maxBy(0)
KeyedStream.maxBy("key")
max 和 maxBy 之間的區別在于 max 返回流中的最大值，但 maxBy 返回具有最大值的鍵， min 和 minBy 同理。
Window
Window 函數允許按時間或其他條件對現有 KeyedStream 進行分組。 以下是以 10 秒的時間窗口聚合：
inputStream.keyBy(0).window(Time.seconds(10));
Flink 定義數據片段以便（可能）處理無限數據流。 這些切片稱為窗口。 此切片有助于通過應用轉換處理數據塊。 要對流進行窗口化，我們需要分配一個可以進行分發的鍵和一個描述要對窗口化流執行哪些轉換的函數
要將流切片到窗口，我們可以使用 Flink 自帶的窗口分配器。 我們有選項，如 tumbling windows, sliding windows, global 和 session windows。 Flink 還允許您通過擴展 WindowAssginer 類來編寫自定義窗口分配器。 這里先預留下篇文章來講解這些不同的 windows 是如何工作的。
WindowAll
windowAll 函數允許對常規數據流進行分組。 通常，這是非并行數據轉換，因為它在非分區數據流上運行。
與常規數據流功能類似，我們也有窗口數據流功能。 唯一的區別是它們處理窗口數據流。 所以窗口縮小就像 Reduce 函數一樣，Window fold 就像 Fold 函數一樣，并且還有聚合。
inputStream.keyBy(0).windowAll(Time.seconds(10));
Union
Union 函數將兩個或多個數據流結合在一起。 這樣就可以并行地組合數據流。 如果我們將一個流與自身組合，那么它會輸出每個記錄兩次。
inputStream.union(inputStream1, inputStream2, ...);
Window join
我們可以通過一些 key 將同一個 window 的兩個數據流 join 起來。
inputStream.join(inputStream1)
.where(0).equalTo(1)
.window(Time.seconds(5))
.apply (new JoinFunction () {...});
以上示例是在 5 秒的窗口中連接兩個流，其中第一個流的第一個屬性的連接條件等于另一個流的第二個屬性。
Split
此功能根據條件將流拆分為兩個或多個流。 當您獲得混合流并且您可能希望多帶帶處理每個數據流時，可以使用此方法。
SplitStream split = inputStream.split(new OutputSelector() {
@Override
public Iterable select(Integer value) {
List output = new ArrayList();
if (value % 2 == 0) {
output.add("even");
}
else {
output.add("odd");
}
return output;
}
});
Select
此功能允許您從拆分流中選擇特定流。
SplitStream split;
DataStream even = split.select("even");
DataStream odd = split.select("odd");
DataStream all = split.select("even","odd");
Project
Project 函數允許您從事件流中選擇屬性子集，并僅將所選元素發送到下一個處理流。
DataStream> in = // [...]
DataStream> out = in.project(3,2);
上述函數從給定記錄中選擇屬性號 2 和 3。 以下是示例輸入和輸出記錄：
(1,10.0,A,B)=> (B,A)
(2,20.0,C,D)=> (D,C)
最后
本文主要介紹了 Flink Data 的常用轉換方式：Map、FlatMap、Filter、KeyBy、Reduce、Fold、Aggregations、Window、WindowAll、Union、Window Join、Split、Select、Project 等。并用了點簡單的 demo 介紹了如何使用，具體在項目中該如何將數據流轉換成我們想要的格式，還需要根據實際情況對待。
關注我
轉載請務必注明原創地址為：http://www.54tianzhisheng.cn/2018/11/04/Flink-Data-transformation/
微信公眾號：zhisheng
另外我自己整理了些 Flink 的學習資料，目前已經全部放到微信公眾號了。你可以加我的微信：zhisheng_tian，然后回復關鍵字：Flink 即可無條件獲取到。
Github 代碼倉庫
https://github.com/zhisheng17/flink-learning/
以后這個項目的所有代碼都將放在這個倉庫里，包含了自己學習 flink 的一些 demo 和博客
相關文章
1、《從0到1學習Flink》—— Apache Flink 介紹
2、《從0到1學習Flink》—— Mac 上搭建 Flink 1.6.0 環境并構建運行簡單程序入門
3、《從0到1學習Flink》—— Flink 配置文件詳解
4、《從0到1學習Flink》—— Data Source 介紹
5、《從0到1學習Flink》—— 如何自定義 Data Source ？
6、《從0到1學習Flink》—— Data Sink 介紹
7、《從0到1學習Flink》—— 如何自定義 Data Sink ？
8、《從0到1學習Flink》—— Flink Data transformation(轉換)
9、《從0到1學習Flink》—— 介紹Flink中的Stream Windows
10、《從0到1學習Flink》—— Flink 中的幾種 Time 詳解
11、《從0到1學習Flink》—— Flink 寫入數據到 ElasticSearch
12、《從0到1學習Flink》—— Flink 項目如何運行？
13、《從0到1學習Flink》—— Flink 寫入數據到 Kafka
