// 캐싱 없이: 비효율적
Dataset<Row> processed = df
    .filter(col("status").equalTo("ACTIVE"))
    .groupBy("category")
    .agg(sum("amount").alias("total"));

// 같은 데이터를 3번 처리 (각각 전체 재계산)
long count = processed.count();              // Job 1: 전체 계산
processed.show();                            // Job 2: 전체 재계산
processed.write().parquet("output");         // Job 3: 전체 재계산

// 캐싱 사용: 효율적
processed.cache();                           // 캐시 등록
long count = processed.count();              // Job 1: 계산 + 캐싱
processed.show();                            // 캐시에서 읽음 (빠름)
processed.write().parquet("output");         // 캐시에서 읽음 (빠름)
processed.unpersist();                       // 캐시 해제

Dataset<Row> df = spark.read().parquet("large-data.parquet");

// cache() 호출 - 기본 스토리지 레벨로 캐시
df.cache();

// 첫 Action에서 실제 캐싱 발생
df.count();  // 데이터 처리 + 메모리에 캐시

// 이후 Action은 캐시에서 읽음
df.filter(col("x").gt(10)).count();
df.groupBy("category").count().show();

// 캐시 해제
df.unpersist();

import org.apache.spark.storage.StorageLevel;

// 메모리에만 캐시 (기본값)
df.persist(StorageLevel.MEMORY_ONLY());

// 메모리 + 디스크 (메모리 부족 시 디스크 사용)
df.persist(StorageLevel.MEMORY_AND_DISK());

// 직렬화하여 메모리에 저장 (메모리 절약, CPU 사용)
df.persist(StorageLevel.MEMORY_ONLY_SER());

// 직렬화 + 디스크 백업
df.persist(StorageLevel.MEMORY_AND_DISK_SER());

// 디스크에만 저장
df.persist(StorageLevel.DISK_ONLY());

// 복제본 2개 유지 (장애 대비)
df.persist(StorageLevel.MEMORY_AND_DISK_2());

df.cache();

// 체크포인트 디렉토리 설정
spark.sparkContext().setCheckpointDir("hdfs:///checkpoints");

// 체크포인트 저장
df.checkpoint();
// 또는 eager checkpoint (즉시 저장)
df.checkpoint(true);

// 단순 반복 사용 → cache
Dataset<Row> frequently = df.filter(...).cache();
for (int i = 0; i < 10; i++) {
    process(frequently);
}

// 긴 lineage 끊기 → checkpoint
Dataset<Row> stage1 = df.map(...).filter(...).join(...);
stage1 = stage1.checkpoint();  // lineage 끊음

Dataset<Row> stage2 = stage1.map(...).filter(...).join(...);
stage2 = stage2.checkpoint();  // 또 끊음

// 머신러닝 반복 알고리즘 → 둘 다 사용
Dataset<Row> data = loadData().cache();  // 반복 접근

for (int iter = 0; iter < 100; iter++) {
    data = processIteration(data);
    if (iter % 10 == 0) {
        data = data.checkpoint();  // 주기적으로 lineage 끊음
    }
}

// 캐시 여부 확인
boolean isCached = spark.catalog().isCached("table_name");

// Spark UI에서 확인
// http://localhost:4040 → Storage 탭

// 특정 DataFrame 해제
df.unpersist();

// 블로킹 방식 (해제 완료까지 대기)
df.unpersist(true);

// 테이블 캐시 해제
spark.catalog().uncacheTable("table_name");

// 모든 캐시 해제
spark.catalog().clearCache();

// 캐시된 테이블 새로고침 (원본 변경 반영)
spark.catalog().refreshTable("table_name");

-- 테이블 캐시
CACHE TABLE employees;

-- 레이지 캐시 (첫 쿼리 시 캐싱)
CACHE LAZY TABLE employees;

-- 쿼리 결과 캐시
CACHE TABLE active_employees AS
SELECT * FROM employees WHERE status = 'ACTIVE';

-- 캐시 해제
UNCACHE TABLE employees;

-- 모든 캐시 해제
CLEAR CACHE;

// 좋음: 여러 번 사용
Dataset<Row> base = loadAndProcess().cache();
analyze1(base);
analyze2(base);
analyze3(base);
base.unpersist();

// 나쁨: 한 번만 사용 (불필요한 캐싱)
Dataset<Row> oneTime = loadData().cache();
oneTime.write().parquet("output");  // 한 번만 사용

// 좋음: 비용이 큰 연산 후 캐시
Dataset<Row> expensive = df
    .filter(...)
    .join(otherDf, ...)      // 비용 큰 조인
    .groupBy(...).agg(...)   // 비용 큰 집계
    .cache();                // 여기서 캐시

// 나쁨: 너무 일찍 캐시 (불필요한 데이터 저장)
Dataset<Row> early = df.cache();  // 아직 필터링 전
Dataset<Row> filtered = early.filter(col("needed").equalTo(true));

// Executor 메모리의 약 60%가 저장/캐시에 사용 (기본)
spark.conf().set("spark.memory.storageFraction", "0.5");  // 50%로 조정

// 캐시 강제 삭제 방지 (메모리 부족 시 캐시 유지)
spark.conf().set("spark.memory.storageFraction", "0.6");

Dataset<Row> cached = expensive.cache();
try {
    // 캐시된 데이터 사용
    doAnalysis(cached);
} finally {
    cached.unpersist();  // 반드시 해제
}

// 큰 객체가 많은 경우 직렬화가 효율적
Dataset<Row> largeObjects = df.persist(StorageLevel.MEMORY_ONLY_SER());

// Kryo 직렬화 사용 (더 효율적)
spark.conf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
spark.conf().set("spark.kryo.registrationRequired", "false");

// 캐시 전 파티션 조정 권장
Dataset<Row> optimized = df
    .filter(...)
    .repartition(100)  // 적절한 파티션 수로 조정
    .cache();

// 파티션이 너무 많으면 메모리 오버헤드
// 파티션이 너무 적으면 병렬성 저하

public class MLPipelineWithCaching {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("ML Pipeline")
                .master("local[*]")
                .getOrCreate();

        // 1. 데이터 로드 및 전처리 (한 번만)
        Dataset<Row> rawData = spark.read()
                .option("header", "true")
                .option("inferSchema", "true")
                .csv("training-data.csv");

        Dataset<Row> processed = rawData
                .na().fill(0)
                .filter(col("label").isNotNull())
                .withColumn("features", createFeatures());

        // 2. 전처리된 데이터 캐시 (반복 학습에 사용)
        processed.cache();
        System.out.println("데이터 캐싱 완료: " + processed.count() + " 행");

        // 3. 학습/검증 분할
        Dataset<Row>[] splits = processed.randomSplit(new double[]{0.8, 0.2});
        Dataset<Row> training = splits[0].cache();
        Dataset<Row> validation = splits[1].cache();

        // 4. 하이퍼파라미터 튜닝 (캐시된 데이터 반복 사용)
        double[] learningRates = {0.1, 0.01, 0.001};
        int[] maxDepths = {5, 10, 15};

        for (double lr : learningRates) {
            for (int depth : maxDepths) {
                // 학습 (캐시에서 읽음 - 빠름)
                trainModel(training, lr, depth);

                // 검증 (캐시에서 읽음 - 빠름)
                double score = evaluate(validation);
                System.out.printf("lr=%.3f, depth=%d, score=%.4f%n", lr, depth, score);
            }
        }

        // 5. 캐시 해제
        training.unpersist();
        validation.unpersist();
        processed.unpersist();

        spark.stop();
    }
}

// 1. 직렬화 사용
df.persist(StorageLevel.MEMORY_AND_DISK_SER());

// 2. 디스크 사용
df.persist(StorageLevel.DISK_ONLY());

// 3. Executor 메모리 증가
spark.conf().set("spark.executor.memory", "8g");

// 4. 불필요한 캐시 해제
anotherDf.unpersist();

// 주의: cache() 후 새 DataFrame이 생성되면 캐시 적용 안 됨
Dataset<Row> df = spark.read().parquet("data");
df.cache();
Dataset<Row> filtered = df.filter(...);  // 새 DataFrame
// filtered는 캐시 안 됨!

// 올바른 방법
Dataset<Row> df = spark.read().parquet("data").cache();
Dataset<Row> filtered = df.filter(...);  // df가 캐시됨