같은 계산을 두 번 하지 않게 — 캐시가 계획에 나타나는 모습

목표

같은 DataFrame 에 행동을 두 번 부를 때 캐시가 없으면 원본부터 다시 계산하는 것을 계획으로 확인하고, cache·persist·unpersist·체크포인트·CACHE TABLE 이 계획과 이벤트 로그에 어떻게 나타나는지 본다. 캐시가 실제로 메모리를 얼마나 먹는지도 블록 갱신 이벤트로 잰다.

왜 중요한가

DataFrame 은 결과가 아니라 만드는 방법(계보)이다. 행동을 부를 때마다 Spark 는 그 방법을 처음부터 다시 실행한다. 같은 중간 결과를 두 번 쓰는 코드는 원본을 두 번 읽고 조인을 두 번 한다 — 오류도 경고도 없이.
cache() 는 그 중간 결과를 실행기에 남겨 두라는 표시다. 표시일 뿐이라 첫 행동이 채우고, 그 뒤의 행동은 계획에서 원본을 읽는 대신 InMemoryTableScan 으로 캐시를 읽는다. 저장 수준(persist)은 메모리·디스크·직렬화 여부를 고르는 것이고, 메모리가 모자라면 무엇을 포기할지를 정한다. 다 쓴 캐시는 unpersist 로 내려놓아야 다른 작업의 메모리가 된다.
계보가 너무 길어지면(반복 계산) 계획 자체가 커져 드라이버가 느려진다. 체크포인트는 결과를 파일로 쓰고 계보를 끊어, 그 뒤의 계획을 '이 파일을 읽는다' 한 줄로 만든다.

단계

1. /root/spk/cache/common.py 에 결제 완료 주문에 금액을 붙이는 함수와 행동 두 번(count, sum(amount))을 부르는 함수를 두고, /root/spk/cache/none.py(앱 spk-cache-none)로 캐시 없이 두 행동의 결과를 /root/spk/cache/out/none.json 에 쓰세요.
2. /root/spk/cache/cached.py(앱 spk-cache-on)에서 같은 DataFrame 에 cache() 를 걸고 두 행동의 결과를 /root/spk/cache/out/cached.json 에, 저장 수준(str(df.storageLevel))을 /root/spk/cache/out/cached_level.txt 에 쓰세요.
3. /root/spk/cache/disk.py(앱 spk-cache-disk)에서 persist(StorageLevel.DISK_ONLY) 로 같은 일을 하고 결과를 /root/spk/cache/out/disk.json, 저장 수준을 /root/spk/cache/out/level.txt 에 쓰세요.
4. /root/spk/cache/unpersist.py(앱 spk-cache-un)에서 캐시 → 행동 → unpersist() → 행동 순서로 돌리고, 캐시 여부(df.is_cached)의 앞뒤를 /root/spk/cache/out/unpersist.json 에 {"before": 참거짓, "after": 참거짓} 로 쓰세요.
5. /root/spk/cache/ckpt.py(앱 spk-cache-ckpt)에서 체크포인트 폴더를 /root/spk/cache/ckpt 로 두고, 금액에 1–10 을 차례로 더하는 열 번의 withColumn 뒤 checkpoint() 한 결과로 채널별 금액 합을 /root/spk/cache/out/ckpt_result 에 Parquet(channel,amount)으로 쓰세요.
6. /root/spk/cache/cache_sql.py(앱 spk-cache-sql)에서 임시 뷰 paid 를 만들고 CACHE TABLE paid_cached AS SELECT channel, amount FROM paid 뒤 채널별 합을 구해 /root/spk/cache/out/cache_table.json 에 {"is_cached": 참거짓, "by_channel": {채널: 합}} 으로 쓰세요.
7. /root/spk/cache/size.py(앱 spk-cache-size, spark.eventLog.logBlockUpdates.enabled=true)로 캐시를 채운 뒤, 그 로그의 rdd_ 블록 크기를 더해 /root/spk/cache/out/cache_size.json 에 {"blocks": 정수, "memory_bytes": 정수, "disk_bytes": 정수} 로 쓰세요.
8. /root/spk/cache/report.md 에 ## 다시 계산하는 비용 ## 저장 수준 ## 캐시의 크기 세 절을 쓰세요. 셋째 절에는 7단계의 블록 수와 메모리 바이트를 넣으세요.

참고

• 스크립트는 /root/spk/cache 에 두고 거기서 돌리세요(from common import …). 원본은 /data/shop/orders.csv·/data/shop/products.csv, 금액 = qty × price.
• 캐시를 읽는 실행의 계획에는 InMemoryTableScan 과 그 아래 InMemoryRelation … StorageLevel(…) 이 보입니다. 원본 스캔(Scan csv)은 InMemoryRelation 설명 안에 남아 있지만 실행되지는 않습니다.
• 블록 갱신 이벤트(SparkListenerBlockUpdated)는 같은 블록이 여러 번 나올 수 있습니다. 블록 ID 마다 마지막 것만 남겨 더하세요.
• 흔한 실수: cache() 만 하고 행동을 안 불러 캐시가 비어 있는 것, 캐시를 다 쓰고 내려놓지 않는 것, 한 앱 안에서 원본과 파생 DataFrame 을 둘 다 캐시해 어느 캐시를 읽었는지 헷갈리는 것.
• 공식 문서: [RDD Persistence](https://spark.apache.org/docs/4.2.0/rdd-programming-guide.html#rdd-persistence) · [Caching Data](https://spark.apache.org/docs/4.2.0/sql-performance-tuning.html#caching-data) · [CACHE TABLE](https://spark.apache.org/docs/4.2.0/sql-ref-syntax-aux-cache-cache-table.html) · [Configuration — spark.eventLog.logBlockUpdates.enabled](https://spark.apache.org/docs/4.2.0/configuration.html)

8단계

1. 캐시 없이 행동 둘 — 원본을 두 번 읽는다
2. cache() — 첫 행동이 채우고 다음 행동이 읽는다
3. persist(DISK_ONLY) — 저장 수준 고르기
4. unpersist — 캐시를 내려놓으면 다시 계산한다
5. 체크포인트로 계보 끊기
6. SQL 의 CACHE TABLE 은 게으르지 않다
7. 캐시가 실제로 먹는 메모리 재기
8. 캐시를 걸 자리와 내려놓을 자리