硬核！一文學完Flink流計算常用算子（Flink算子大全）

回覆“資料”領取獨家整理的學習資料！

直入正題！

Flink和Spark類似，也是一種一站式處理的框架；既可以進行批處理（DataSet），也可以進行實時處理（DataStream）。

所以下面將Flink的算子分為兩大類：一類是DataSet，一類是DataStream。

DataSet

一、Source算子

1. fromCollection

fromCollection：從本地集合讀取數據

例：

<code>val env = ExecutionEnvironment.getExecutionEnvironment
val textDataSet: DataSet[
String
] = env.fromCollection(
List(
"1,張三"
, 
"2,李四"
, 
"3,王五"
, 
"4,趙六"
)
)
/<code>

2. readTextFile

readTextFile：從文件中讀取：

<code>val textDataSet: DataSet[
String
] = env.readTextFile(
"/data/a.txt"
)
/<code>

3. readTextFile：遍歷目錄

readTextFile可以對一個文件目錄內的所有文件，包括所有子目錄中的所有文件的遍歷訪問方式：

<code>val parameters = 
new
 Configuration
//
 recursive.file.enumeration 開啟遞歸
parameters.setBoolean(
"recursive.file.enumeration"
, 
true
)
val file = env.readTextFile(
"/data"
).withParameters(parameters)
/<code>

4. readTextFile：讀取壓縮文件

對於以下壓縮類型，不需要指定任何額外的inputformat方法，flink可以自動識別並且解壓。但是，壓縮文件可能不會並行讀取，可能是順序讀取的，這樣可能會影響作業的可伸縮性。

壓縮方法文件擴展名是否可並行讀取DEFLATE.deflatenoGZip.gz .gzipnoBzip2.bz2noXZ.xzno

<code>
val
 file = env.readTextFile(
"/data/file.gz"
)
/<code>

二、Transform轉換算子

因為Transform算子基於Source算子操作，所以首先構建Flink執行環境及Source算子，後續Transform算子操作基於此：

<code>val env = ExecutionEnvironment.getExecutionEnvironment
val textDataSet: DataSet[
String
] = env.fromCollection(
List(
"張三,1"
, 
"李四,2"
, 
"王五,3"
, 
"張三,4"
)
)
/<code>

1. map

將DataSet中的每一個元素轉換為另外一個元素：

<code>// 使用
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

/<code>

2. flatMap

將DataSet中的每一個元素轉換為0...n個元素：

<code>/
/ 使用flatMap操作，將集合中的數據：

/
/ 根據第一個元素，進行分組

/
/ 根據第二個元素，進行聚合求值

 

val result = textDataSet.flatMap(line => line)

.groupBy(0) /
/ 根據第一個元素，進行分組

.sum(1) /
/ 根據第二個元素，進行聚合求值


result.print

/<code>

3. mapPartition

將一個分區中的元素轉換為另一個元素：

<code>
//
 
使用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

/<code>

4. filter

過濾出來一些符合條件的元素，返回boolean值為true的元素：

<code>val source: DataSet[
String
] = env.fromElements( 
"java"
, 
"scala"
, 
"java"
)
val filter:DataSet[
String
] = source.filter(
line
 =>
 line.contains(
"java"
))

/<code>

5. reduce

可以對一個dataset或者一個group來進行聚合計算，最終聚合成一個元素：

<code>
//
 使用 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

/<code>

6. reduceGroup

將一個dataset或者一個group聚合成一個或多個元素

。

reduceGroup是reduce的一種優化方案；

它會先分組reduce，然後在做整體的reduce；這樣做的好處就是可以減少網絡IO：

<code>
//
 使用 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

/<code>

7. minBy和maxBy

選擇具有最小值或最大值的元素：

<code>// 使用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) // 每個人的最小值
/<code>

8. Aggregate

在數據集上進行聚合求最值（最大值、最小值）：

<code>
val
 
data
 = new mutable.MutableList[(
Int
, String, 
Double
)]
data
.+=((
1
, 
"yuwen"
, 
89.0
))
data
.+=((
2
,  
"shuxue"
, 
92.2
))
data
.+=((
3
, 
"yuwen"
, 
89.99
))








/<code>

Aggregate只能作用於元組上

注意：

要使用aggregate，只能使用字段索引名或索引名稱來進行分組 <code>groupBy(0)/<code>，否則會報一下錯誤:

Exception in thread "main" java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.

9. distinct

去除重複的數據：

<code>// 數據源使用上一題的
// 使用distinct操作，根據科目去除集合中重複的元組數據 


val value: DataSet[(Int, String, Double)] = 
input
.distinct(
1
)
value.
print

/<code>

10. first

取前N個數：

<code>
input
.first
(
2
) 
/<code>

11. join

將兩個DataSet按照一定條件連接到一起，形成新的DataSet：

<code>/
/ 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)

}

/<code>

12. leftOuterJoin

左外連接,左邊的Dataset中的每一個元素，去連接右邊的元素

此外還有：

rightOuterJoin：右外連接,左邊的Dataset中的每一個元素，去連接左邊的元素

fullOuterJoin：全外連接,左右兩邊的元素，全部連接

下面以 leftOuterJoin 進行示例：

<code> 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==
){

(
first._1,first._2,""
)

} 
else
{

(
first._1,first._2,second._2
)

}

}
).
print

/<code>

13. cross

交叉操作，通過形成這個數據集和其他數據集的笛卡爾積，創建一個新的數據集

和join類似，但是這種交叉操作會產生笛卡爾積，在數據比較大的時候，是非常消耗內存的操作：

<code>val cross = input1.cross(input2){
(input1 , input2) => (input1._1,input1._2,input1._3,input2._2)
}

cross.
print 

/<code>

14. union

聯合操作，創建包含來自該數據集和其他數據集的元素的新數據集,不會去重：

<code>val unionData: DataSet[
String
] = elements1
.
union
(
elements2
)
.
union
(
elements3
)


/<code>

15. rebalance

Flink也有數據傾斜的時候，比如當前有數據量大概10億條數據需要處理，在處理過程中可能會發生如圖所示的狀況：

這個時候本來總體數據量只需要10分鐘解決的問題，出現了數據傾斜，機器1上的任務需要4個小時才能完成，那麼其他3臺機器執行完畢也要等待機器1執行完畢後才算整體將任務完成；所以在實際的工作中，出現這種情況比較好的解決方案就是接下來要介紹的—rebalance（內部使用round robin方法將數據均勻打散。這對於數據傾斜時是很好的選擇。）

<code>

/<code>

16. partitionByHash

按照指定的key進行hash分區：

<code>
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
/<code>

17. partitionByRange

根據指定的key對數據集進行範圍分區：

<code>
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
/<code>

18. sortPartition

根據指定的字段值進行分區的排序：

<code>
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)




/<code>

三、Sink算子

1. collect

將數據輸出到本地集合：

<code>
result
.collect

/<code>

2. writeAsText

將數據輸出到文件

Flink支持多種存儲設備上的文件，包括本地文件，hdfs文件等

Flink支持多種文件的存儲格式，包括text文件，CSV文件等

<code>/
/ 將數據寫入本地文件 

result.writeAsText("/data
/a", WriteMode.OVERWRITE)


/
/ 將數據寫入HDFS

result.writeAsText("hdfs:/
/node01:9000/data
/a", WriteMode.OVERWRITE)

/<code>

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， 先引入依賴：

<code>
<
dependency
>

 
<
groupId
>
org.apache.flink
groupId
>

<
artifactId
>
flink-connector-kafka-0.11_2.11
artifactId
>

<
version
>
1.10.0
version
>

dependency
>

/<code>

將Kafka數據寫入Flink：

<code>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))
/<code>

基於網絡套接字的：

<code>val 
source
 = env.socketTextStream(
"IP"
, PORT)
/<code>

二、Transform轉換算子

1. map

將DataSet中的每一個元素轉換為另外一個元素：

<code>dataStream.map {  
x
 =>
 x * 
2
 }
/<code>

2. FlatMap

採用一個數據元並生成零個，一個或多個數據元。將句子分割為單詞的flatmap函數：

<code>dataStream.flatMap { 
str
 => 
str
.split(
" "
) }
/<code>

3. Filter

計算每個數據元的布爾函數，並保存函數返回true的數據元。過濾掉零值的過濾器：

<code>dataStream.
filter
 { 
_
 != 
0
 }
/<code>

4. KeyBy

邏輯上將流分區為不相交的分區。具有相同Keys的所有記錄都分配給同一分區。在內部，keyBy是使用散列分區實現的。指定鍵有不同的方法。

此轉換返回KeyedStream，其中包括使用被Keys化狀態所需的KeyedStream：

<code>
dataStream
.keyBy
(0) 
/<code>

5. Reduce

被Keys化數據流上的“滾動”Reduce。將當前數據元與最後一個Reduce的值組合併發出新值：

<code>keyedStream.
reduce
 { 
_
 + 
_
 } 
/<code>

6. Fold

具有初始值的被Keys化數據流上的“滾動”摺疊。將當前數據元與最後摺疊的值組合併發出新值：

<code>val result: DataStream[
String
] = keyedStream.fold(
"start"
)(
(
str, i
) =>
 { str + 
"-"
 + i }) 


/<code>

7. Aggregations

在被Keys化數據流上滾動聚合。min和minBy之間的差異是min返回最小值，而minBy返回該字段中具有最小值的數據元（max和maxBy相同）：

<code>
keyedStream
.sum
(0);

keyedStream
.min
(0);

keyedStream
.max
(0);

keyedStream
.minBy 
(0);

keyedStream
.maxBy
(0);
/<code>

8. Window

可以在已經分區的KeyedStream上定義Windows。Windows根據某些特徵（例如，在最後5秒內到達的數據）對每個Keys中的數據進行分組。這裡不再對窗口進行詳解，有關窗口的完整說明，請查看這篇文章：Flink 中極其重要的 Time 與 Window 詳細解析

<code>
dataStream
.keyBy
(0)
.window
(
TumblingEventTimeWindows
.of
(
Time
.seconds
(5))); 
/<code>

9. WindowAll

Windows可以在常規DataStream上定義。Windows根據某些特徵（例如，在最後5秒內到達的數據）對所有流事件進行分組。

注意：在許多情況下，這是非並行轉換。所有記錄將收集在windowAll 算子的一個任務中。

<code>
dataStream
.windowAll
(
TumblingEventTimeWindows
.of
(
Time
.seconds
(5)))
/<code>

10. Window Apply

將一般函數應用於整個窗口。

注意：如果您正在使用windowAll轉換，則需要使用AllWindowFunction。

下面是一個手動求和窗口數據元的函數：

<code>windowedStream.apply { WindowFunction }

allWindowedStream.apply { AllWindowFunction }
/<code>

11. Window Reduce

將函數縮減函數應用於窗口並返回縮小的值：

<code>windowedStream.
reduce
 { 
_
 + 
_
 }
/<code>

12. Window Fold

將函數摺疊函數應用於窗口並返回摺疊值：

<code>val result: DataStream[
String
] = windowedStream.fold(
"start"
, (
str
, i) => { 
str
 + 
"-"
 + i }) 


/<code>

13. Union

兩個或多個數據流的聯合，創建包含來自所有流的所有數據元的新流。注意：如果將數據流與自身聯合，則會在結果流中獲取兩次數據元：

<code>dataStream 
.
union
(
otherStream1
, otherStream2, ...)
/<code>

14. Window Join

在給定Keys和公共窗口上連接兩個數據流：

<code>
dataStream
.join
(otherStream)
.where
()
.equalTo
()
.window
(TumblingEventTimeWindows.of(Time.seconds(
3
)))
.apply
 (new JoinFunction  {...})
/<code>

15. Interval Join

在給定的時間間隔內使用公共Keys關聯兩個被Key化的數據流的兩個數據元e1和e2，以便e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound

<code>
am
.intervalJoin
(otherKeyedStream)
.between
(Time.milliseconds(-
2
), Time.milliseconds(
2
))
.upperBoundExclusive
(true)
.lowerBoundExclusive
(true)
.process
(new IntervalJoinFunction {...})
/<code>

16. Window CoGroup

在給定Keys和公共窗口上對兩個數據流進行Cogroup：

<code>
dataStream
.coGroup
(otherStream)
.where
(
0
)
.equalTo
(
1
)
.window
(TumblingEventTimeWindows.of(Time.seconds( 
3
)))
.apply
 (new CoGroupFunction  {...})
/<code>

17. Connect

“連接”兩個保存其類型的數據流。連接允許兩個流之間的共享狀態：

<code>DataStream someStream = ... DataStream<
String
> otherStream = ... ConnectedStreamsString> connectedStreams = someStream.connect(otherStream)


/<code>

18. CoMap，CoFlatMap

類似於連接數據流上的map和flatMap：

<code>connectedStreams.map(
(_ : Int)
 =>
 
true
,
(_ : String)
 =>
 
false
)connectedStreams.flatMap( 

(_ : Int)
 =>
 
true
,
(_ : String)
 =>
 
false
)
/<code>

19. Split

根據某些標準將流拆分為兩個或更多個流：

<code>val split = someDataStream.split(
(num: Int) =>
(num % 
2
) match {
case
 
0
 => 
List
(
"even"
)
case
 
1
 => 
List
(
"odd"
)
})
/<code> 

20. Select

從拆分流中選擇一個或多個流：

<code>SplitStream 
split
;DataStream even = 
split
.select(
"even"
);DataStream odd = 
split
.select(
"odd"
);DataStream all = 
split
.select(
"even"
,
"odd"
)
/<code>

三、Sink算子

支持將數據輸出到：

• 本地文件(參考批處理)

• 本地集合(參考批處理)

• HDFS(參考批處理)

除此之外，還支持：

• sink到kafka

• sink到mysql

• sink到redis

下面以sink到kafka為例：

<code>val sinkTopic = "test"

//樣例類
case class Student(id: Int, name: String, addr: String, sex: String)
val mapper: ObjectMapper = new ObjectMapper

//將對象轉換成字符串
deftoJsonString(T: Object): String = {
mapper.registerModule(DefaultScalaModule)
mapper.writeValueAsString(T)
}

defmain(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")
}/<code>

最後說一句(求關注，別白嫖我)

掃一掃，我們的故事就開始了。

另外公眾號改變了推送規則，大家看文章不要忘記點擊最下方的在看，點贊按鈕，這樣微信自動識別為常看公眾號，否則很可能推送的文章可能淹沒在別的文章找不到，謝謝大家。

想加好友的加一下啊，微信號：create17_ ，多交朋友，才能走的更遠。

讓我知道你在看