Apache Spark — 느린 잡의 답은 실행 계획과 이벤트 로그에 있다 · 캐시와 재계산 · 이론

DataFrame 은 결과가 아니라 조리법이라 매번 다시 끓인다

한 줄 요약

DataFrame 은 계산된 결과가 아니라 원본에서 결과까지 가는 계보(lineage) 이므로 행동을 부를 때마다 원본부터 다시 계산한다. 캐시는 첫 계산의 결과를 붙잡아 두는 것이고, 체크포인트는 결과를 파일로 쓰고 계보를 끊는 것이다.

개념 지도: 원본에서 결과까지 가는 계보(lineage) · 두 번 · 중간 결과는 버린다. · 여러 번 쓰는 중간 결과만 골라서

왜 같은 계산을 두 번 하나

CSV 를 읽어 정제하고 조인한 DataFrame 하나로 count() 를 부르고, 이어서 같은 것을 파일로 쓴다고 하자. 코드만 보면 정제는 한 번이다. 실제로는 두 번 일어난다. 이벤트 로그를 보면 두 잡이 모두 CSV 스캔에서 시작한다.

이유는 변환이 게으르기 때문이다. filter 나 join 은 계획에 한 줄을 더할 뿐 아무것도 계산하지 않는다. 행동이 불리면 Spark 는 그 계획을 원본부터 끝까지 실행하고, 결과를 드라이버나 파일로 넘긴 뒤 중간 결과는 버린다. 다음 행동은 다시 원본에서 시작한다. 이것은 결함이 아니라 설계다. 중간 결과를 전부 붙잡아 두면 메모리가 금방 차고, 계보만 있으면 잃어버린 조각은 언제든 다시 만들 수 있다.

그래서 여러 번 쓰는 중간 결과만 골라서 붙잡는 일은 사람의 몫이다. 그것이 캐시다.

어떻게 동작하나

cache() 도 게으르다. 부르는 순간에는 "이 결과를 처음 계산할 때 붙잡아 두라" 는 표시만 남는다. 첫 행동이 돌 때 각 파티션이 계산되면서 저장되고, 그 뒤의 행동은 계획에서 원본 스캔 대신 InMemoryTableScan 으로 시작한다. [RDD 프로그래밍 가이드](https://spark.apache.org/docs/4.2.0/rdd-programming-guide.html)의 설명대로 처음 계산될 때 저장되고, 그 뒤의 행동이 그것을 재사용한다.

from pyspark import StorageLevelclean = raw.filter("qty > 0").join(products, "product_id")clean.cache()                      # 표시만 한다 — 아직 아무것도 저장되지 않았다clean.count()                      # 첫 행동: 계산하면서 저장clean.write.parquet("/tmp/out")    # 계획이 InMemoryTableScan 에서 시작한다clean.unpersist()                  # 다음 행동은 다시 원본부터slim = raw.select("order_id", "qty").persist(StorageLevel.DISK_ONLY)

저장 수준이 어디에 두느냐를 정한다. 헷갈리기 쉬운 곳이 기본값이다. RDD 의 cache() 는 메모리에만 두는 MEMORY_ONLY 다. 반면 [DataFrame 의 cache()](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.cache.html)는 MEMORY_AND_DISK_DESER 로, 메모리에 들어가지 않는 파티션은 디스크에 둔다. 문서는 이 기본이 3.0 에서 스칼라에 맞춰 바뀌었다고 적는다. 다른 수준을 원하면 persist() 에 넘기는데, 이미 저장 수준이 정해진 DataFrame 에는 새 수준을 줄 수 없다. 먼저 unpersist() 해야 한다.

DataFrame 캐시는 행 그대로가 아니라 칼럼 형식으로 메모리에 놓인다. [성능 조정 문서](https://spark.apache.org/docs/4.2.0/sql-performance-tuning.html)는 캐시된 표에서 필요한 칼럼만 읽고 압축을 스스로 맞춘다고 설명한다.

어느 수준을 고를지에 대해 RDD 가이드의 조언은 짧다. 메모리에 편하게 들어가면 기본으로 두고, 계산이 비싸거나 걸러 내는 양이 많을 때가 아니면 디스크로 흘리지 말라. 다시 계산하는 것이 디스크에서 읽는 것만큼 빠를 수 있기 때문이다. 원본이 로컬 CSV 한 개이고 변환이 가벼우면 DISK_ONLY 캐시는 이득이 거의 없다.

캐시가 사는 곳 — 통합 메모리

캐시는 공짜 공간에 놓이지 않는다. [튜닝 가이드](https://spark.apache.org/docs/4.2.0/tuning.html)에 따르면 셔플·조인·정렬·집계가 쓰는 실행 메모리와 캐시가 쓰는 저장 메모리는 한 영역을 나눠 쓴다. 그 영역의 크기가 spark.memory.fraction(기본 0.6)이고, JVM 힙에서 300MiB 를 뺀 나머지에 대한 비율이다. 그 안에서 spark.memory.storageFraction(기본 0.5)만큼은 실행 메모리가 캐시를 밀어내지 못하는 몫이다.

이 실습의 드라이버 메모리 1g 로 어림하면 (1024 - 300) × 0.6, 즉 400MiB 남짓을 실행과 캐시가 나눠 쓴다. 큰 표를 캐시하면 그만큼 조인과 정렬이 쓸 자리가 줄고, 캐시가 넘치면 RDD 가이드의 말대로 가장 오래 쓰지 않은 파티션부터(LRU) 밀려난다. 밀려난 파티션은 다음에 필요할 때 계보를 따라 다시 계산된다. 그래서 캐시는 틀린 답을 내지 않는다. 느려질 뿐이다.

unpersist 와 체크포인트

다 쓴 캐시는 unpersist() 로 놓아 준다. RDD 가이드는 이 호출이 기본으로 기다리지 않는다고 적는다. 자원이 실제로 풀릴 때까지 기다리려면 blocking 인자를 준다. 놓아 준 뒤의 행동은 계획에서 다시 원본 스캔으로 시작한다. 또 하나, DataFrame 캐시는 클러스터의 모든 세션이 공유한다. 같은 계획을 다른 세션이 캐시해 두었으면 내 쿼리도 그것을 쓴다.

캐시는 계보를 그대로 둔다. 잃은 파티션을 다시 만들 수 있는 것도 그 덕이다. 그런데 반복 알고리즘처럼 같은 DataFrame 에 변환을 수십 번 덧대면 계보 자체가 문제가 된다. [체크포인트 문서](https://spark.apache.org/docs/4.2.0/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.checkpoint.html)는 이런 경우 계획이 기하급수로 자랄 수 있고, 체크포인트가 논리 계획을 잘라 낸다고 설명한다. 결과를 SparkContext.setCheckpointDir() 나 spark.checkpoint.dir 로 정한 디렉터리에 파일로 쓰고, 그 뒤의 계획은 원본이 아니라 그 파일에서 시작한다. 계획에는 원본 스캔 대신 이미 있는 RDD 를 읽는 노드가 나타난다. 기본은 즉시(eager) 실행이라 부르는 순간 잡이 돈다.

localCheckpoint() 는 파일 대신 실행기의 캐시 체계에 저장한다. 빠르지만 문서가 직접 말하듯 믿을 수 없다 — 실행기를 잃으면 계보도 끊겨 있어 다시 만들 길이 없다.

SQL 에도 같은 도구가 있다. [CACHE TABLE](https://spark.apache.org/docs/4.2.0/sql-ref-syntax-aux-cache-cache-table.html) 은 DataFrame 의 cache() 와 달리 기본이 즉시이고, LAZY 를 붙여야 처음 쓸 때 캐시한다. 저장 수준을 따로 주지 않으면 MEMORY_AND_DISK 다.

현장에서 만나는 모습

첫째, 한 번만 쓰는 것을 캐시한다. 행동이 하나뿐인 DataFrame 을 캐시하면 저장하는 비용만 더 든다. 캐시는 두 번 이상 쓰는 갈림길에만 둔다.

둘째, 캐시를 붙잡아 두고 잊는다. 긴 노트북이나 서비스 안에서 캐시가 쌓이면 실행 메모리가 줄어 셔플이 디스크로 흘러넘치기 시작한다. 쓰고 나면 놓아 준다.

셋째, 캐시한 줄 알았는데 계획에 없다. cache() 뒤에 새 변환을 붙인 다른 DataFrame 에 행동을 부르면, 그 DataFrame 의 계획 안에서 캐시된 부분만 재사용된다. 캐시가 정말 쓰였는지는 행동 뒤 explain() 에서 InMemoryTableScan 을 찾아 확인한다. [웹 UI 의 Storage 탭](https://spark.apache.org/docs/4.2.0/web-ui.html)도 실체화된 뒤에야 캐시를 보여 준다.

실무에서 진짜 중요한 것

DataFrame 은 조리법이다. 행동마다 원본부터 다시 계산한다.
cache() 는 게으르다. 첫 행동이 돌 때 저장된다.
DataFrame 의 기본 저장 수준은 MEMORY_AND_DISK_DESER, RDD 는 MEMORY_ONLY 다.
캐시는 실행 메모리와 같은 영역을 나눠 쓴다. 다 쓰면 unpersist 한다.
계보가 너무 길면 체크포인트로 끊는다. 캐시는 계보를 끊지 않는다.

다음 실습에서 할 것

캐시 없이 두 행동을 부르고 두 잡이 모두 CSV 를 다시 읽는 것을 이벤트 로그와 계획으로 확인한다. cache() 뒤에는 두 번째 행동의 계획이 InMemoryTableScan 에서 시작하는 것과 저장 수준을 보고, persist 로 디스크에만 두는 수준을 지정해 본다. unpersist 앞뒤로 캐시 여부가 바뀌는 것을 확인하고, 변환을 열 번 덧댄 DataFrame 을 체크포인트로 끊어 같은 답이 나오는지 본 뒤 SQL 의 CACHE TABLE 로 같은 일을 해 본다. 마지막으로 블록 갱신 기록을 켠 앱의 이벤트 로그에서 캐시 블록의 크기를 더해 캐시가 메모리를 얼마나 차지했는지 숫자로 잰다.