Flink WordCount

일괄 처리
파일 내용:

hello world
hello scala
hello flink

코드:

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

object WordCount {

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

    //            
    val env = ExecutionEnvironment.getExecutionEnvironment

    val inputDataSet = env.readTextFile("D:\\project\\idea\\FlinkTutorial\\src\\main\\resources\\hello.txt")

    val wordCountDataSet = inputDataSet
      .flatMap(_.split(" "))
      .map((_,1))
      .groupBy(0)//    0     
      .sum(1)//    1     

    wordCountDataSet.print()

  }

}

　　
이류 처리

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

object StreamWordCount {

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

    //            
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //  socket   
    val textDataStream = env.socketTextStream("hadoop102", 7777)

    val wordCountDataStream = textDataStream
      .flatMap(_.split("\\s"))
      .map((_, 1))
      .keyBy(0)
      .sum(1)

    wordCountDataStream.print()

    //    
    env.execute("   ")
  }

}