[Spark] Spark 3.x Structured Streaming의 checkpoint는 k8s환경에서 어떻게 관리해야할까?

Spark 3.x로 올라오면서 Structured Streaming(이하 s.stream)이 지원되면서 group by, window 등 강력하면서도 다양한 기능을 streaming 에서도 사용할 수 있게 되었다.

기존 Discretized Streaming(이하 d.stream)과 다른 점 중 두드러지는 변화는 바로 checkpoint를 저장한다는 것이다.

checkpoint란?

streaming의 metadata 또는 kafka offset 등을 저장하는 데이터로 streaming app의 진행상황을 기록해둘 수 있다.

streaming app의 진행상황을 기록해두면, app이 종료되더라도 해당 지점부터 재수행이 가능하기 때문이다.

동작 원리나 여러가지 부가적인 기능 및 내용들이 많이 있지만, 자세한 것은 공식문서를 참고하길 바란다.

(안타깝게도 언제나 그렇듯 공식문서의 설명은 충분하지 않긴 하다)

공식문서: https://spark.apache.org/docs/latest/streaming-programming-guide.html#checkpointing

checkpoint 지정 방법

간단하다. SparkContext 또는 DataStreamWriter의 option을 주면 된다.

// 1. SparkContext에 지정하는 방법
val sparkConf = new SparkConf().setAppName("Checkpoint Example")
val checkpointDirectory: String = "/my/directory/checkpoint"

val ssc = new StreamingContext(sparkConf, Seconds(1))  
ssc.checkpoint(checkpointDirectory)
...


// 2. DataStreamWriter의 option으로 주는 방법
val dataset: Dataset[Schema] = getDatasetExample()
val checkpointDirectory: String = "/my/directory/checkpoint"

dataset.writeStream
    .option("checkpointLocation", checkpointDirectory)
    .trigger(ProcessingTime("10 seconds")
    .foreachBatch(saveFunction _)
    .queryName("Checkpoint Example")
    .start()

그렇다면 checkpoint는 어디다가 저장하는 것이 좋을까?

만약 물리장비에서 spark application을 운영한다면(아마도 spark on yarn), local directory에 저장하면 된다.
제일 간편하고, 성능이슈도 없다.

문제는 k8s환경에서(spark on k8s) 운영할 때인데, k8s환경의 local에 checkpoint를 저장하게 되면 pod이 내려갔을 때 checkpoint 정보를 모두 잃고 pod이 내려갔다가 올라오는 동안의 데이터가 유실되게 된다.

1. hdfs

hdfs는 좋은 선택지다.

그러나 secure hdfs를 사용하는 경우 i/o 성능이 뒷받침되지 않으면서 walCommit 시간이 늘어나서, micro batch 간격이 5~10초인 경우 lag이 점점 증가하는 경우가 발생할 수 있다.
(walCommit관련 내용은 추후에 또 다뤄보겠다)

secure hdfs를 사용하지 않는 경우 보안 이슈가 발생할 수 있으니 추천하기는 어렵다.

2. fuse

fuse는 사용하기는 편하겠지만 여전히 성능상의 이슈를 가지고 있다.
여러 pod에서 mount하기 때문에 write할 때 일종의 mutex를 획득해야 하는 경우가 많은데, 이 때 i/o 성능이 눈에 띄게 하락하게 된다.

역시나 walCommit 시간이 증가하는 이슈가 발생한다.

3. rbd

rbd는 거의 정답에 가까운 선택지다.
i/o 성능이 준수하고 옵션을 통해 persistence까지 보장이 가능하다. 필자도 현재 rbd를 checkpoint를 저장하는 용도로 사용하고 있으며, 아직 이슈가 된 적이 없다.

k8s환경에서 s.stream app을 운영하게 된다면, 성능 및 보안 측면에서 흠잡을 데 없는 rbd를 checkpoint directory로 사용하는 것을 강력하게 추천한다.