Flink 시작하기 #3 Word Count Sample (Scala)

Flink를 사용해 가장 기본 예제가 되는 Word Count 를 만들어보자. Scala, SBT를 사용한 예제이다.

우선 Dependency 다음과 같다.

libraryDependencies += "org.apache.flink" % "flink-streaming-scala_2.11" % "1.1.3"

그리고 코드를 작성 해 보자 

import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.api.windowing.time.Time object WordCount { def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment val text = env.socketTextStream("localhost", 9999) val count = text .flatMap(_.split(" ")) .map((_, 1)) .keyBy(0) .timeWindow(Time.seconds(5), Time.seconds(1)) .sum(1) count.print().setParallelism(1) env.execute("Socket WordCount") } }

netcat으로 입력받은 Text line을 input 으로 받는다. 문장을 단어 별로 분리한 후 5초간의 데이터를 1초 마다 수행하여 각 단어별 카운팅을 한다.(Windows 연산 참조: https://ci.apache.org/projects/flink/flink-docs-release-1.1/apis/streaming/windows.html )

netcat을 실행하고 문장을 입력하면 아래와 같이 결과가 나타난다.

(world,1)

(Hello,1)

(world,1)

(Hello,1)

(world,1)

(Hello,1)

(world,1)

(Hello,2)

(Flink,1)

(Hello,2)

(Flink,1)

(world,1)

(Hello,2)

(Scala,1)

(Flink,1)

(Hello,2)

(Scala,1)

(Flink,1)

(Hello,2)

(Scala,1)

(Flink,1)

(Hello,1)

(Scala,1)

(Hello,1)

(Scala,1)

(is,1)

(Streaming,1)

(Flink,1)

(is,1)

(Streaming,1)

(Flink,1)

(is,1)

(Streaming,1)

(Flink,1)

(is,1)

(Streaming,1)

(Flink,1)

(is,1)

(Streaming,1)

(Flink,1)

끝