Flink 시작하기 #7 Batch (DataSet API)

Flink Batch는 streaming API와 거의 유사하다.

기능적으로 Streaming을 처리하는 방식을 지속적인 데이터가 아닌 bounded 데이터를 처리하는 개념이다. 하지만 내부적으로는 fault tolerance에 대한 처리에서 chkeck point가 아닌 재실행하는 방식을 사용하는 등 좀 다르게 처리되고 있지만 API를 사용하는 레벨에서는 알 필요는 없다.

가장 먼저 Stream처리와 다른 부분은 ExecutionEnvironment 를 가져오는 부분이 다르다.

val env = ExecutionEnvironment.getExecutionEnvironment

transformation API는 Streaming에서 사용하는 API와 사용법과 종류가 같다. 물론 스트림 데이터를 처리하는 것이 아니기 때문에 Windows API는 없다.

그리고 source와 Sink 부분은 Streaming에서 기본적으로 제공하는 file, collection은 Dataset에서도 제공하지만 addSource, addSink 처럼 Custom은 지원하지 않는다. 왜 지원안하는지 잘모르겠다. 아무튼 지금 버전인 1.2 버전에서는 사용할 수 없다.

대신 readFile에서 "hdfs://.."로 하둡으로부터 파일을 읽을 수 있다. 그리고 .deflate, gz, gzip, b2, xz 등 압출파일도 지원한다.

간단히 파일을 읽어서 처리후 다시 파일로 저장하는 예제를 작성해보자.

Flink github에 있는 README.md 파일을 word count해서 텍스트 파일로 저장하는 예제이다.

import org.apache.flink.api.scala._ import org.apache.flink.core.fs.FileSystem.WriteMode object WordCount { def main(args: Array[String]): Unit = { val env = ExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) // read from file val text = env.readTextFile("src/main/resources/README_FLINK.md") val count = text .flatMap(_.split(" ")) .map((_, 1)) .groupBy(0) .sum(1) .filter(_._2 > 10) // write to file count.writeAsText("src/main/resources/wordcount.txt", WriteMode.OVERWRITE); env.execute("flink batch example") } }

readTextFile을 통해 local 파일을 가져온다. 그리고 streaming API에서 word count를 작성했던것과 동일하게 작성한다.

그리고 sink 메소드로 writeAsText를 통해 파일로 저장한다. 이때 paralleism(1)로 설정을 해야만 1개에 파일로 저장된다. 기본설정이 4이기 때문에 저장된 파일을 쉽게 확인하기 위해 1로 설정했다.

결과

(,86) (*,20) (Apache,11) (Flink,12) (Scala,11) (and,20) (for,15) (the,32) (to,18)

끝