flink基础wordcount scala版

import org.apache.flink.api.java.utils.ParameterTool

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import org.apache.flink.streaming.api.windowing.time.Time

object FlinkFirstAppScala {

def main(args: Array[String]):Unit = {

// the host and the port to connect to

    var hostname:String ="localhost"

    var port:Int =0

    try {

val params = ParameterTool.fromArgs(args)

hostname = params.get("hostname","192.168.21.138")

port = params.getInt("port",9999)

}catch {

case e:Exception => {

System.err.println("No port specified. Please run 'SocketWindowWordCount " +

"--hostname <hostname> --port <port>', where hostname (localhost by default) and port " +

"is the address of the text server")

System.err.println("To start a simple text server, run 'netcat -l <port>' " +

"and type the input text into the command line")

return

      }

}

// get the execution environment

    val env=StreamExecutionEnvironment.getExecutionEnvironment

    import org.apache.flink.streaming.api.scala._

// get input data by connecting to the socket; parse the data, group it, window it, and aggregate the counts

    env.socketTextStream(hostname,port)

.flatMap(x=>x.split(" "))

.map(x=>WC(x,1))

.keyBy("word")

.timeWindow(Time.seconds(4),Time.seconds(2))

.sum("count")

.print

.setParallelism(1)

/** Data type for words with count */

    env.execute("FlinkFirstAppScala")

}

case class WC(word:String,count:Long)

}