[Spark] Streaming Word Count 샘플 (Scala)

Streaming Word Count 샘플

Spark Streaming 에서 Word Count 샘플 코드를 만들어보자

기존에 배치 프로세싱에서의 Word Counting 참고: http://gyrfalcon.tistory.com/110

Spark Streaming Dependency를 추가한다. Spark core와 동일버젼으로 추가한다.

<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.11</artifactId> <version>2.0.1</version> </dependency>

인풋 소스로 여러 다양한 방법이 있겠지만 가장 기본이 되는 Socket text 방식으로 text를 입력받는 방식을 사용한다.

StreamingContext 으로 Stream Spark Context를 생성한다. 이때 시간 interval을 입력해야하는데 Spark Streaming은 일정시간 주기로 RDD를 수행하는 방식을 사용하기 때문이다.

Context로 SockectTextStream을 생성한다. 그리고 일반 배치 프로세싱과 동일한 코드를 작성하면된다.

마지막 Action에서 reduceByKeyAndWindow 란 메소드를 사용했다. Streaming에 있는 Windows 방식으로 동작하는 메소드이다.

간단히 설명하면 Streming은 일정 주기로 RDD를 실행하는건데 한번에 처리되는 범위를 지정해주는 개념이라고 생각하면된다.

import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} object WordCount { def main(args: Array[String]): Unit = { val conf = new SparkConf().setMaster("local[2]").setAppName("Spark Streaming WordCount") val ssc = new StreamingContext(conf, Seconds(1)) val lines = ssc.socketTextStream("localhost", 9999) val wordCount = lines .flatMap(_.split(" ")) .map((_, 1)) .reduceByKeyAndWindow(_+_, Seconds(5)) wordCount.print() ssc.start() ssc.awaitTermination() } }

TextStream을 생성하기 위해 Console에서 Netcat을 이용해보자

명령어는 nc -lk 9999 로 하면된다.

이제 코드를 실행해 보자

------------------------------------------- Time: 1481161338000 ms ------------------------------------------- (Hello,1) (World,1) ------------------------------------------- Time: 1481161339000 ms ------------------------------------------- (Hello,1) (World,1) ------------------------------------------- Time: 1481161340000 ms ------------------------------------------- (Hello,2) (World,2) ------------------------------------------- Time: 1481161341000 ms ------------------------------------------- (Hello,2) (World,2) ------------------------------------------- Time: 1481161342000 ms ------------------------------------------- (Hello,3) (World,3) ------------------------------------------- Time: 1481161343000 ms ------------------------------------------- (Hello,2) (World,2) (HHHH,1) ------------------------------------------- Time: 1481161344000 ms ------------------------------------------- (Hello,2) (World,2) (HHHH,1) ------------------------------------------- Time: 1481161345000 ms ------------------------------------------- (Hello,1) (World,1) (HHHH,1) ------------------------------------------- Time: 1481161346000 ms ------------------------------------------- (Hello,1) (World,1) (HHHH,1) ------------------------------------------- Time: 1481161347000 ms ------------------------------------------- (HHHH,1) ------------------------------------------- Time: 1481161348000 ms -------------------------------------------

위와 같이 콘솔에 나타날것이다. 

window를 5초로 지정했기 때문에 최근 5초간 입력된 Text에 대해 카운팅이 들어가서 Hello World를 3번만 입력했지만 어려번 화면에 나타나는것이다.

끝