Flink1.7.2源碼分析

時間 2019-11-13

標籤 flink1.7.2 flink 源碼分析简体版

原文原文鏈接

技術交流

Flink版本爲1.7.2

源碼

https://github.com/opensourceteams/flink-maven-scala

中文文檔

Flink 1.7中文文檔(官網英文翻譯過來) 在線版: http://flink.apachecn.org/docs/1.7-SNAPSHOT/#/README
Flink 1.7中文文檔(官網英文翻譯過來) PDF版: https://github.com/opensourceteams/flink-maven-scala/blob/master/doc/flink-1.7-%E4%B8%AD%E6%96%87%E6%96%87%E6%A1%A3(%E5%AE%98%E7%BD%91%E8%8B%B1%E6%96%87%E7%BF%BB%E8%AF%91%E8%BF%87%E6%9D%A5).pdf

Flink 本地運行交互Shell

start-scala-shell.sh local
參數說明: [local | remote <host> <port> | yarn]

benv.fromElements(1,2,3).map(i => i * i ).print

輸出結果

1
4
9

運行 jar 到 Flink 集羣

flink run -c  com.opensourceteams.module.bigdata.flink.example.stream.worldcount.nc.SocketWindowWordCount    ./flink-maven-scala-2-0.0.1.jar

建立flink java 項目

mvn archetype:generate                               \
      -DarchetypeGroupId=org.apache.flink              \
      -DarchetypeArtifactId=flink-quickstart-java     \
      -DarchetypeVersion=1.7.1
      -DgroupId=com.opensourceteams \
      -DartifactId=flink-maven-java \
      -Dversion=0.0.1 \
      -Dpackage=com.opensourceteams.module.bigdata.flink  \
      -DinteractiveMode=false

建立flink scala項目

mvn archetype:generate                               \
      -DarchetypeGroupId=org.apache.flink              \
      -DarchetypeArtifactId=flink-quickstart-scala     \
      -DarchetypeVersion=1.7.1
      -DgroupId=com.opensourceteams \
      -DartifactId=flink-maven-scala-2 \
      -Dversion=0.0.1 \
      -Dpackage=com.opensourceteams.module.bigdata.flink  \
      -DinteractiveMode=false

查看jar中文件列表

jar tvf test.jar

maven 運行某個類

mvn exec:java -Dexec.mainClass=wikiedits.WikipediaAnalysis

執行計劃圖

用Firefox 打開，顯示的比較全(別瀏覽器有顯示不全的現象)
地址:https://flink.apache.org/visualizer

//執行計劃
      //println(env.getExecutionPlan)
      //StreamGraph
     //println(env.getStreamGraph.getStreamingPlanAsJSON)

Execute Plan

{"nodes":[{"id":1,"type":"Source: Socket Stream","pact":"Data Source","contents":"Source: Socket Stream","parallelism":1},{"id":2,"type":"Flat Map","pact":"Operator","contents":"Flat Map","parallelism":1,"predecessors":[{"id":1,"ship_strategy":"FORWARD","side":"second"}]},{"id":3,"type":"Map","pact":"Operator","contents":"Map","parallelism":1,"predecessors":[{"id":2,"ship_strategy":"FORWARD","side":"second"}]},{"id":5,"type":"Window(TumblingProcessingTimeWindows(3000), ProcessingTimeTrigger, SumAggregator, PassThroughWindowFunction)","pact":"Operator","contents":"Window(TumblingProcessingTimeWindows(3000), ProcessingTimeTrigger, SumAggregator, PassThroughWindowFunction)","parallelism":1,"predecessors":[{"id":3,"ship_strategy":"HASH","side":"second"}]},{"id":6,"type":"Sink: Print to Std. Out","pact":"Data Sink","contents":"Sink: Print to Std. Out","parallelism":1,"predecessors":[{"id":5,"ship_strategy":"FORWARD","side":"second"}]}]}

StreamGraph Plan

{"nodes":[{"id":1,"type":"Source: Socket Stream","pact":"Data Source","contents":"Source: Socket Stream","parallelism":1},{"id":2,"type":"Flat Map","pact":"Operator","contents":"Flat Map","parallelism":1,"predecessors":[{"id":1,"ship_strategy":"FORWARD","side":"second"}]},{"id":3,"type":"Map","pact":"Operator","contents":"Map","parallelism":1,"predecessors":[{"id":2,"ship_strategy":"FORWARD","side":"second"}]},{"id":5,"type":"Window(TumblingProcessingTimeWindows(3000), ProcessingTimeTrigger, SumAggregator, PassThroughWindowFunction)","pact":"Operator","contents":"Window(TumblingProcessingTimeWindows(3000), ProcessingTimeTrigger, SumAggregator, PassThroughWindowFunction)","parallelism":1,"predecessors":[{"id":3,"ship_strategy":"HASH","side":"second"}]},{"id":6,"type":"Sink: Print to Std. Out","pact":"Data Sink","contents":"Sink: Print to Std. Out","parallelism":1,"predecessors":[{"id":5,"ship_strategy":"FORWARD","side":"second"}]}]}

Flink 環境，配置

Flink 源碼debug方法: https://github.com/opensourceteams/flink-maven-scala/blob/master/md/flink-debug.md
Flink 名詞術語 : https://github.com/opensourceteams/flink-maven-scala/blob/master/md/flink-concept.md
Flink 源碼編譯: https://github.com/opensourceteams/flink-maven-scala/blob/master/md/flink-build.md

example

scala 版Flink WordCount單詞統計 : https://github.com/opensourceteams/flink-maven-scala/blob/master/md/wordCount-scala.mdjava
wordCount Dataset批處理node
- start-scala-shell.sh local
- 參數說明:[local | remote <host> <port> | yarn]
```
benv.fromElements("a b a c").flatMap(x => x.split(" ")).map((_,1)).groupBy(0).sum(1).print
```
- 輸出結果
```
(a,2)
  (b,1)
  (c,1)
```
Flink 1.7.2 DataStream operator 示例 : https://github.com/opensourceteams/flink-maven-scala/blob/master/md/datastream/flink-datastream-operator.mdgit