大數(shù)據(jù)：Windows下配置flink的Stream

對(duì)于開(kāi)發(fā)人員來(lái)說(shuō)，最希望的是需要在windows中進(jìn)行測(cè)試，然后把調(diào)試好的程序放在集群中運(yùn)行。下面寫(xiě)一個(gè)Socket，上面是監(jiān)控本地的一個(gè)運(yùn)行端口，來(lái)實(shí)時(shí)的提取數(shù)據(jù)。獲取視頻中文檔資料及完整視頻的伙伴請(qǐng)加QQ群：947967114

成都創(chuàng)新互聯(lián)成立于2013年，是專業(yè)互聯(lián)網(wǎng)技術(shù)服務(wù)公司，擁有項(xiàng)目成都網(wǎng)站制作、成都網(wǎng)站設(shè)計(jì)、外貿(mào)營(yíng)銷網(wǎng)站建設(shè)網(wǎng)站策劃，項(xiàng)目實(shí)施與項(xiàng)目整合能力。我們以讓每一個(gè)夢(mèng)想脫穎而出為使命，1280元天河做網(wǎng)站,已為上家服務(wù),為天河各地企業(yè)和個(gè)人服務(wù),聯(lián)系電話:18980820575

下面是一段代碼：

import org.apache.flink.api.java.utils.ParameterToolimport org.apache.flink.streaming.api.scala._import org.apache.flink.streaming.api.windowing.time.Timeobject SocketWindowWordCount { def main(args: Array[String]): Unit = { val port:Int=try{ ParameterTool.fromArgs(args).getInt("port") } catch{ case e:Exception=>{ System.out.print("Noport spacified.Please run 'SocketWindowWordCount'") return } } val env:StreamExecutionEnvironment=StreamExecutionEnvironment.getExecutionEnvironment val text=env.socketTextStream("localhost",port,'\n') val windowWordCount=text.flatMap{w=>w.split("\s+")}.map{w=>WordWithCount(w,1)}. keyBy("word").timeWindow(Time.seconds(5),Time.seconds(1)).sum("count") windowWordCount.print().setParallelism(1) env.execute("Socket Window WordCount") }}case class WordWithCount(word: String, count: Long){}

這個(gè)程序有幾個(gè)需要注意的。第一個(gè)org.apache.flink.streaming.api.windowing.time.Time是這個(gè)時(shí)間包的使用。第二個(gè)是org.apache.flink.streaming.api.scala._，不要直接引入org.apache.flink.streaming.api.scala.StreamExecutionEnvironment。
大數(shù)據(jù)：Windows下配置flink的Stream