硬核！一文學完Flink流計算常用運算元（Flink運算元大全）
直入正題！
Flink和Spark類似，也是一種一站式處理的框架；既可以進行批處理（DataSet），也可以進行實時處理（DataStream）。
所以下面將Flink的運算元分為兩大類：一類是DataSet，一類是DataStream。
DataSet
一、Source運算元
1. fromCollection
fromCollection：從本地集合讀取資料
例：
val env = ExecutionEnvironment。getExecutionEnvironment
val textDataSet： DataSet［String］ = env。fromCollection（
List（“1，張三”， “2，李四”， “3，王五”， “4，趙六”）
）
2. readTextFile
readTextFile：從檔案中讀取
val textDataSet： DataSet［String］ = env。readTextFile（“/data/a。txt”）
3. readTextFile：遍歷目錄
readTextFile可以對一個檔案目錄內的所有檔案，包括所有子目錄中的所有檔案的遍歷訪問方式
val parameters = new Configuration
// recursive。file。enumeration 開啟遞迴
parameters。setBoolean（“recursive。file。enumeration”， true）
val file = env。readTextFile（“/data”）。withParameters（parameters）
4. readTextFile：讀取壓縮檔案
對於以下壓縮型別，不需要指定任何額外的inputformat方法，flink可以自動識別並且解壓。但是，壓縮檔案可能不會並行讀取，可能是順序讀取的，這樣可能會影響作業的可伸縮性。
壓縮方法
副檔名
是否可並行讀取
因格式問題，此處顯示不全，完整的表格內容可到公號【五分鐘學大資料】檢視
val file = env。readTextFile（“/data/file。gz”）
二、Transform轉換運算元
因為Transform運算元基於Source運算元操作，所以首先構建Flink執行環境及Source運算元，後續Transform運算元操作基於此：
val env = ExecutionEnvironment。getExecutionEnvironment
val textDataSet： DataSet［String］ = env。fromCollection（
List（“張三，1”， “李四，2”， “王五，3”， “張三，4”）
）
1. map
將DataSet中的每一個元素轉換為另外一個元素
// 使用map將List轉換為一個Scala的樣例類
case class User（name： String， id： String）
val userDataSet： DataSet［User］ = textDataSet。map {
text =>
val fieldArr = text。split（“，”）
User（fieldArr（0）， fieldArr（1））
}
userDataSet。print（）
2. flatMap
將DataSet中的每一個元素轉換為0。。。n個元素。
// 使用flatMap操作，將集合中的資料：
// 根據第一個元素，進行分組
// 根據第二個元素，進行聚合求值
val result = textDataSet。flatMap（line => line）
。groupBy（0） // 根據第一個元素，進行分組
。sum（1） // 根據第二個元素，進行聚合求值
result。print（）
3. mapPartition
將一個分割槽中的元素轉換為另一個元素
// 使用mapPartition操作，將List轉換為一個scala的樣例類
case class User（name： String， id： String）
val result： DataSet［User］ = textDataSet。mapPartition（line => {
line。map（index => User（index。_1， index。_2））
}）
result。print（）
4. filter
過濾出來一些符合條件的元素，返回
boolean值為true
的元素
val source： DataSet［String］ = env。fromElements（“java”， “scala”， “java”）
val filter：DataSet［String］ = source。filter（line => line。contains（“java”））//過濾出帶java的資料
filter。print（）
5. reduce
可以對一個dataset或者一個group來進行聚合計算，最終
聚合成一個元素
// 使用 fromElements 構建資料來源
val source = env。fromElements（（“java”， 1）， （“scala”， 1）， （“java”， 1））
// 使用map轉換成DataSet元組
val mapData： DataSet［（String， Int）］ = source。map（line => line）
// 根據首個元素分組
val groupData = mapData。groupBy（_。_1）
// 使用reduce聚合
val reduceData = groupData。reduce（（x， y） => （x。_1， x。_2 + y。_2））
// 列印測試
reduceData。print（）
6. reduceGroup
將一個dataset或者一個group
聚合成一個或多個元素
。
reduceGroup是reduce的一種最佳化方案；
它會先分組reduce，然後在做整體的reduce；這樣做的好處就是可以減少網路IO
// 使用 fromElements 構建資料來源
val source： DataSet［（String， Int）］ = env。fromElements（（“java”， 1）， （“scala”， 1）， （“java”， 1））
// 根據首個元素分組
val groupData = source。groupBy（_。_1）
// 使用reduceGroup聚合
val result： DataSet［（String， Int）］ = groupData。reduceGroup {
（in： Iterator［（String， Int）］， out： Collector［（String， Int）］） =>
val tuple = in。reduce（（x， y） => （x。_1， x。_2 + y。_2））
out。collect（tuple）
}
// 列印測試
result。print（）
7. minBy和maxBy
選擇具有最小值或最大值的
元素
// 使用minBy操作，求List中每個人的最小值
// List（“張三，1”， “李四，2”， “王五，3”， “張三，4”）
case class User（name： String， id： String）
// 將List轉換為一個scala的樣例類
val text： DataSet［User］ = textDataSet。mapPartition（line => {
line。map（index => User（index。_1， index。_2））
}）
val result = text
。groupBy（0） // 按照姓名分組
。minBy（1） // 每個人的最小值
8. Aggregate
在資料集上進行聚合求
最值
（最大值、最小值）
val data = new mutable。MutableList［（Int， String， Double）］
data。+=（（1， “yuwen”， 89。0））
data。+=（（2， “shuxue”， 92。2））
data。+=（（3， “yuwen”， 89。99））
// 使用 fromElements 構建資料來源
val input： DataSet［（Int， String， Double）］ = env。fromCollection（data）
// 使用group執行分組操作
val value = input。groupBy（1）
// 使用aggregate求最大值元素
。aggregate（Aggregations。MAX， 2）
// 列印測試
value。print（）
Aggregate只能作用於元組上
注意：
要使用aggregate，只能使用欄位索引名或索引名稱來進行分組
groupBy（0）
，否則會報一下錯誤：
Exception in thread “main” java。lang。UnsupportedOperationException： Aggregate does not support grouping with KeySelector functions， yet。
9. distinct
去除重複的資料
// 資料來源使用上一題的
// 使用distinct操作，根據科目去除集合中重複的元組資料
val value： DataSet［（Int， String， Double）］ = input。distinct（1）
value。print（）
10. first
取前N個數
input。first（2） // 取前兩個數
11. join
將兩個DataSet按照一定條件連線到一起，形成新的DataSet
// s1 和 s2 資料集格式如下：
// DataSet［（Int， String，String， Double）］
val joinData = s1。join（s2） // s1資料集 join s2資料集
。where（0）。equalTo（0） { // join的條件
（s1， s2） => （s1。_1， s1。_2， s2。_2， s1。_3）
}
12. leftOuterJoin
左外連線，左邊的Dataset中的每一個元素，去連線右邊的元素
此外還有：
rightOuterJoin：右外連線，左邊的Dataset中的每一個元素，去連線左邊的元素
fullOuterJoin：全外連線，左右兩邊的元素，全部連線
下面以 leftOuterJoin 進行示例：
val data1 = ListBuffer［Tuple2［Int，String］］（）
data1。append（（1，“zhangsan”））
data1。append（（2，“lisi”））
data1。append（（3，“wangwu”））
data1。append（（4，“zhaoliu”））
val data2 = ListBuffer［Tuple2［Int，String］］（）
data2。append（（1，“beijing”））
data2。append（（2，“shanghai”））
data2。append（（4，“guangzhou”））
val text1 = env。fromCollection（data1）
val text2 = env。fromCollection（data2）
text1。leftOuterJoin（text2）。where（0）。equalTo（0）。apply（（first，second）=>{
if（second==null）{
（first。_1，first。_2，“null”）
}else{
（first。_1，first。_2，second。_2）
}
}）。print（）
13. cross
交叉操作，透過形成這個資料集和其他資料集的笛卡爾積，建立一個新的資料集
和join類似，但是這種交叉操作會產生笛卡爾積，在
資料比較大的時候，是非常消耗記憶體的操作
val cross = input1。cross（input2）{
（input1 ， input2） => （input1。_1，input1。_2，input1。_3，input2。_2）
}
cross。print（）
14. union
聯合操作，建立包含來自該資料集和其他資料集的元素的新資料集，
不會去重
val unionData： DataSet［String］ = elements1。union（elements2）。union（elements3）
// 去除重複資料
val value = unionData。distinct（line => line）
15. rebalance
Flink也有資料傾斜的時候，比如當前有資料量大概10億條資料需要處理，在處理過程中可能會發生如圖所示的狀況：
這個時候本來總體資料量只需要10分鐘解決的問題，出現了資料傾斜，機器1上的任務需要4個小時才能完成，那麼其他3臺機器執行完畢也要等待機器1執行完畢後才算整體將任務完成； 所以在實際的工作中，出現這種情況比較好的解決方案就是接下來要介紹的—
rebalance
（內部使用round robin方法將資料均勻打散。這對於資料傾斜時是很好的選擇。）
// 使用rebalance操作，避免資料傾斜
val rebalance = filterData。rebalance（）
16. partitionByHash
按照指定的key進行hash分割槽
val data = new mutable。MutableList［（Int， Long， String）］
data。+=（（1， 1L， “Hi”））
data。+=（（2， 2L， “Hello”））
data。+=（（3， 2L， “Hello world”））
val collection = env。fromCollection（data）
val unique = collection。partitionByHash（1）。mapPartition{
line =>
line。map（x => （x。_1 ， x。_2 ， x。_3））
}
unique。writeAsText（“hashPartition”， WriteMode。NO_OVERWRITE）
env。execute（）
17. partitionByRange
根據指定的key對資料集進行範圍分割槽
val data = new mutable。MutableList［（Int， Long， String）］
data。+=（（1， 1L， “Hi”））
data。+=（（2， 2L， “Hello”））
data。+=（（3， 2L， “Hello world”））
data。+=（（4， 3L， “Hello world， how are you？”））
val collection = env。fromCollection（data）
val unique = collection。partitionByRange（x => x。_1）。mapPartition（line => line。map{
x=>
（x。_1 ， x。_2 ， x。_3）
}）
unique。writeAsText（“rangePartition”， WriteMode。OVERWRITE）
env。execute（）
18. sortPartition
根據指定的欄位值進行分割槽的排序
val data = new mutable。MutableList［（Int， Long， String）］
data。+=（（1， 1L， “Hi”））
data。+=（（2， 2L， “Hello”））
data。+=（（3， 2L， “Hello world”））
data。+=（（4， 3L， “Hello world， how are you？”））
val ds = env。fromCollection（data）
val result = ds
。map { x => x }。setParallelism（2）
。sortPartition（1， Order。DESCENDING）//第一個引數代表按照哪個欄位進行分割槽
。mapPartition（line => line）
。collect（）
println（result）
三、Sink運算元
1. collect
將資料輸出到本地集合
result。collect（）
2. writeAsText
將資料輸出到檔案
Flink支援多種儲存裝置上的檔案，包括本地檔案，hdfs檔案等
Flink支援多種檔案的儲存格式，包括text檔案，CSV檔案等
// 將資料寫入本地檔案
result。writeAsText（“/data/a”， WriteMode。OVERWRITE）
// 將資料寫入HDFS
result。writeAsText（“hdfs：//node01：9000/data/a”， WriteMode。OVERWRITE）
DataStream
和DataSet一樣，DataStream也包括一系列的Transformation操作
一、Source運算元
Flink可以使用 StreamExecutionEnvironment。addSource（source） 來為我們的程式新增資料來源。
Flink 已經提供了若干實現好了的 source functions，當然我們也可以透過實現 SourceFunction 來自定義非並行的source或者實現 ParallelSourceFunction 介面或者擴充套件 RichParallelSourceFunction 來自定義並行的 source。
Flink在流處理上的source和在批處理上的source基本一致。大致有4大類：
基於
本地集合
的source（Collection-based-source）
基於
檔案
的source（File-based-source）- 讀取文字檔案，即符合 TextInputFormat 規範的檔案，並將其作為字串返回
基於
網路套接字
的source（Socket-based-source）- 從 socket 讀取。元素可以用分隔符切分。
自定義
的source（Custom-source）
下面使用addSource將Kafka資料寫入Flink為例：
如果需要外部資料來源對接，可使用addSource，如將Kafka資料寫入Flink， 先引入依賴：
<！—— https：//mvnrepository。com/artifact/org。apache。flink/flink-connector-kafka-0。11 ——>
將Kafka資料寫入Flink：
val properties = new Properties（）
properties。setProperty（“bootstrap。servers”， “localhost：9092”）
properties。setProperty（“group。id”， “consumer-group”）
properties。setProperty（“key。deserializer”， “org。apache。kafka。common。serialization。StringDeserializer”）
properties。setProperty（“value。deserializer”， “org。apache。kafka。common。serialization。StringDeserializer”）
properties。setProperty（“auto。offset。reset”， “latest”）
val source = env。addSource（new FlinkKafkaConsumer011［String］（“sensor”， new SimpleStringSchema（）， properties））
基於網路套接字的：
val source = env。socketTextStream（“IP”， PORT）
二、Transform轉換運算元
1. map
將DataSet中的每一個元素轉換為另外一個元素
dataStream。map { x => x * 2 }
2. FlatMap
採用一個數據元並生成零個，一個或多個數據元。將句子分割為單詞的flatmap函式
dataStream。flatMap { str => str。split（“ ”） }
3. Filter
計算每個資料元的布林函式，並儲存函式返回true的資料元。過濾掉零值的過濾器
dataStream。filter { _ ！= 0 }
4. KeyBy
邏輯上將流分割槽為不相交的分割槽。具有相同Keys的所有記錄都分配給同一分割槽。在內部，keyBy（）是使用雜湊分割槽實現的。指定鍵有不同的方法。
此轉換返回KeyedStream，其中包括使用被Keys化狀態所需的KeyedStream。
dataStream。keyBy（0）
5. Reduce
被Keys化資料流上的“滾動”Reduce。將當前資料元與最後一個Reduce的值組合併發出新值
keyedStream。reduce { _ + _ }
6. Fold
具有初始值的被Keys化資料流上的“滾動”摺疊。將當前資料元與最後摺疊的值組合併發出新值
val result： DataStream［String］ = keyedStream。fold（“start”）（（str， i） => { str + “-” + i }）
// 解釋：當上述程式碼應用於序列（1，2，3，4，5）時，輸出結果“start-1”，“start-1-2”，“start-1-2-3”，。。。
7. Aggregations
在被Keys化資料流上滾動聚合。min和minBy之間的差異是min返回最小值，而minBy返回該欄位中具有最小值的資料元（max和maxBy相同）。
keyedStream。sum（0）；
keyedStream。min（0）；
keyedStream。max（0）；
keyedStream。minBy（0）；
keyedStream。maxBy（0）；
8. Window
可以在已經分割槽的KeyedStream上定義Windows。Windows根據某些特徵（例如，在最後5秒內到達的資料）對每個Keys中的資料進行分組。這裡不再對視窗進行詳解，有關視窗的完整說明，請檢視這篇文章：
Flink 中極其重要的 Time 與 Window 詳細解析
dataStream。keyBy（0）。window（TumblingEventTimeWindows。of（Time。seconds（5）））；
9. WindowAll
Windows可以在常規DataStream上定義。Windows根據某些特徵（例如，在最後5秒內到達的資料）對所有流事件進行分組。
注意：在許多情況下，這是非並行轉換。所有記錄將收集在windowAll 運算元的一個任務中。
dataStream。windowAll（TumblingEventTimeWindows。of（Time。seconds（5）））
10. Window Apply
將一般函式應用於整個視窗。
注意：如果您正在使用windowAll轉換，則需要使用AllWindowFunction。
下面是一個手動求和視窗資料元的函式
windowedStream。apply { WindowFunction }
allWindowedStream。apply { AllWindowFunction }
11. Window Reduce
將函式縮減函式應用於視窗並返回縮小的值
windowedStream。reduce { _ + _ }
12. Window Fold
將函式摺疊函式應用於視窗並返回摺疊值
val result： DataStream［String］ = windowedStream。fold（“start”， （str， i） => { str + “-” + i }）
// 上述程式碼應用於序列（1，2，3，4，5）時，將序列摺疊為字串“start-1-2-3-4-5”
13. Union
兩個或多個數據流的聯合，建立包含來自所有流的所有資料元的新流。注意：如果將資料流與自身聯合，則會在結果流中獲取兩次資料元
dataStream。union（otherStream1， otherStream2， 。。。）
14. Window Join
在給定Keys和公共視窗上連線兩個資料流
dataStream。join（otherStream）
。where（
。window（TumblingEventTimeWindows。of（Time。seconds（3）））
。apply （new JoinFunction （） {。。。}）
15. Interval Join
在給定的時間間隔內使用公共Keys關聯兩個被Key化的資料流的兩個資料元e1和e2，以便e1。timestamp + lowerBound <= e2。timestamp <= e1。timestamp + upperBound
am。intervalJoin（otherKeyedStream）
。between（Time。milliseconds（-2）， Time。milliseconds（2））
。upperBoundExclusive（true）
。lowerBoundExclusive（true）
。process（new IntervalJoinFunction（） {。。。}）
16. Window CoGroup
在給定Keys和公共視窗上對兩個資料流進行Cogroup
dataStream。coGroup（otherStream）
。where（0）。equalTo（1）
。window（TumblingEventTimeWindows。of（Time。seconds（3）））
。apply （new CoGroupFunction （） {。。。}）
17. Connect
“連線”兩個儲存其型別的資料流。連線允許兩個流之間的共享狀態
DataStream
// 。。。 代表省略中間操作
18. CoMap，CoFlatMap
類似於連線資料流上的map和flatMap
connectedStreams。map（
（_ ： Int） => true，
（_ ： String） => false）connectedStreams。flatMap（
（_ ： Int） => true，
（_ ： String） => false）
19. Split
根據某些標準將流拆分為兩個或更多個流
val split = someDataStream。split（
（num： Int） =>
（num % 2） match {
case 0 => List（“even”）
case 1 => List（“odd”）
}）
20. Select
從拆分流中選擇一個或多個流
SplitStream
三、Sink運算元
支援將資料輸出到：
本地檔案（參考批處理）
本地集合（參考批處理）
HDFS（參考批處理）
除此之外，還支援：
sink到kafka
sink到mysql
sink到redis
下面以sink到kafka為例：
val sinkTopic = “test”
//樣例類
case class Student（id： Int， name： String， addr： String， sex： String）
val mapper： ObjectMapper = new ObjectMapper（）
//將物件轉換成字串
def toJsonString（T： Object）： String = {
mapper。registerModule（DefaultScalaModule）
mapper。writeValueAsString（T）
}
def main（args： Array［String］）： Unit = {
//1。建立流執行環境
val env = StreamExecutionEnvironment。getExecutionEnvironment
//2。準備資料
val dataStream： DataStream［Student］ = env。fromElements（
Student（8， “xiaoming”， “beijing biejing”， “female”）
）
//將student轉換成字串
val studentStream： DataStream［String］ = dataStream。map（student =>
toJsonString（student） // 這裡需要顯示SerializerFeature中的某一個，否則會報同時匹配兩個方法的錯誤
）
//studentStream。print（）
val prop = new Properties（）
prop。setProperty（“bootstrap。servers”， “node01：9092”）
val myProducer = new FlinkKafkaProducer011［String］（sinkTopic， new KeyedSerializationSchemaWrapper［String］（new SimpleStringSchema（））， prop）
studentStream。addSink（myProducer）
studentStream。print（）
env。execute（“Flink add sink”）
}
——end——