###########################################實驗步驟######################################

①啟動集群

$FLINK_HOME/bin/start-cluster.sh

②開啟一個socket

nc -lk 9999

mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 muamua mua2 mua3 mua2 mua
mua mua2 mua3 mua2 mua

③編譯命令:

mvn scala:compile package

④提交flink任務(wù)，產(chǎn)生另外一個socket

flink run -c WordCount target/scala-module-dependency-sample-1.0-SNAPSHOT.jar --port 9999

?

然后上述兩個socket就會通過tcp進(jìn)行通信

###########################實驗結(jié)果###################################################

輸出結(jié)果在:

$FLINK_HOME/log/flink-appleyuchi-taskexecutor-2-Desktop.out

3> WordWithCount(mua3,2)
4> WordWithCount(mua2,4)
4> WordWithCount(mua2,14)
3> WordWithCount(mua3,7)
3> WordWithCount(mua3,12)
4> WordWithCount(mua2,24)
3> WordWithCount(muamua,1)
3> WordWithCount(muamua,1)
4> WordWithCount(mua2,24)
3> WordWithCount(mua3,12)
3> WordWithCount(muamua,1)
4> WordWithCount(mua2,24)
3> WordWithCount(mua3,12)
3> WordWithCount(muamua,1)
4> WordWithCount(mua2,20)
3> WordWithCount(mua3,10)
3> WordWithCount(muamua,1)
4> WordWithCount(mua2,10)
3> WordWithCount(mua3,5)

#####################################附錄-項目結(jié)構(gòu)##################################

├── pom.xml
├── src
│???└── main
│??? ? ?└── scala
│??? ? ? ? ?└── WordCount.scala
└── 運行方法.txt

#########################附錄-代碼#######################################

完整的pom.xml:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"><modelVersion>4.0.0</modelVersion><groupId>sample</groupId><artifactId>scala-module-dependency-sample</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>2.11.8</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-scala_2.11</artifactId><version>1.6.2</version></dependency><!-- <dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.11</artifactId><version>1.6.2</version></dependency>--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.11</artifactId><version>1.6.2</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.11</artifactId><version>1.6.2</version></dependency></dependencies><build><plugins><plugin><groupId>org.scala-tools</groupId><artifactId>maven-scala-plugin</artifactId><version>2.15.2</version><executions><execution><id>scala-compile-first</id><goals><goal>compile</goal></goals><configuration><includes><include>**/*.scala</include></includes></configuration></execution><execution><id>scala-test-compile</id><goals><goal>testCompile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build></project>

WordCount.scala

import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.api.windowing.time.Timeobject WordCount {// Data type for words with countcase class WordWithCount(word: String, count: Long)def main(args: Array[String]): Unit = {// 獲取執(zhí)行器的環(huán)境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//獲取數(shù)據(jù): 從socket中獲取val textDataStream = env.socketTextStream("Desktop", 9999, '\n')val tupDataStream = textDataStream.flatMap(_.split(" ")).map(WordWithCount(_,1))//groupby: 按照指定的字段聚合val windowDstram = tupDataStream.keyBy("word").timeWindow(Time.seconds(5), Time.seconds(1))//窗口bsize=5秒, slid=1swindowDstram.sum("count").print()//啟動執(zhí)行器，執(zhí)行程序env.execute("Socket Window WordCount")} }

####################################附錄-排查######################################################

Flink運行實驗后找不到輸出結(jié)果的原因有如下幾種：

①前面的job沒有cancel影響下一個,可以輸入flink list后用flink cancel刪除。

②yarn的queue資源被占滿了。(flink on yarn模式)

③ 輸出結(jié)果在yarn的界面中的某個節(jié)點或者某個節(jié)點的#$FLINK_HOME/log下面

④不要刪除$FLINK/log下面的flink-用戶名-taskexecutor-2-Desktop.out，這個不會在新job時自動生成，只會在start-all.sh啟動時生成，如果刪除了，除非重啟flink集群，否則是看不到實驗結(jié)果了。

⑤啟動集群前6123端口沒有關(guān)閉，導(dǎo)致重啟后，任務(wù)無法提交

Caused by: java.net.BindException: Could not start actor system on any port in port range 6123

注意提交任務(wù)前，首先要確保$FLINK_HOME/log下面輸入grep -ri error沒有error

?

?

?

總結(jié)

以上是生活随笔為你收集整理的flink的scala版本的wordcount+flink没有输出结果的几种原因的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。