# Spark 버전 확인
spark-submit --version

# Spark UI 접근 확인 (애플리케이션 실행 중일 때)
curl -s http://localhost:4040/api/v1/applications | head -1

Dataset<Row> result = df
    .groupBy("category")
    .agg(count("*"), sum("amount"))
    .orderBy(col("count").desc());

// 실행 계획 확인
result.explain();

// 셔플 메트릭 로깅
df.write()
    .mode("overwrite")
    .parquet("output");

// Spark UI → Stages → 완료된 Stage → Shuffle Read/Write 확인

// 두 번의 groupBy = 두 번의 셔플
Dataset<Row> counts = df.groupBy("category").count();
Dataset<Row> sums = df.groupBy("category").agg(sum("amount"));
Dataset<Row> result = counts.join(sums, "category");  // 또 셔플!

// 하나의 groupBy에서 모든 집계 수행
Dataset<Row> result = df.groupBy("category")
    .agg(
        count("*").alias("count"),
        sum("amount").alias("total_amount"),
        avg("amount").alias("avg_amount")
    );

Dataset<Row> joined = large.join(small, "key");
Dataset<Row> result = joined.filter(col("status").equalTo("ACTIVE"));

// 데이터량을 먼저 줄여서 셔플 크기 감소
Dataset<Row> filteredLarge = large.filter(col("status").equalTo("ACTIVE"));
Dataset<Row> result = filteredLarge.join(small, "key");

Dataset<Row> unique = df.distinct();  // 전체 컬럼 기준 distinct

// 필요한 컬럼만 선택 후 distinct (셔플 데이터량 감소)
Dataset<Row> unique = df.select("id", "category").distinct();

import static org.apache.spark.sql.functions.broadcast;

// 작은 테이블(수십 MB 이하)을 모든 Executor에 복제
Dataset<Row> result = largeTable.join(
    broadcast(smallTable),
    "key"
);

SparkSession spark = SparkSession.builder()
    // 100MB 이하 테이블 자동 브로드캐스트
    .config("spark.sql.autoBroadcastJoinThreshold", "104857600")
    .getOrCreate();

result.explain();
// BroadcastHashJoin이 보이면 셔플 없이 조인

SparkSession spark = SparkSession.builder()
    // 기본값 200, 데이터 크기에 따라 조정
    .config("spark.sql.shuffle.partitions", "100")
    .getOrCreate();

SparkSession spark = SparkSession.builder()
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.minPartitionSize", "64MB")
    .getOrCreate();

// 조인 키 기준으로 버케팅
df.write()
    .bucketBy(100, "user_id")  // 100개 버킷
    .sortBy("user_id")
    .saveAsTable("users_bucketed");

transactions.write()
    .bucketBy(100, "user_id")  // 동일한 버킷 수
    .sortBy("user_id")
    .saveAsTable("transactions_bucketed");

Dataset<Row> users = spark.table("users_bucketed");
Dataset<Row> transactions = spark.table("transactions_bucketed");

// 같은 버킷 수 + 같은 키 = 셔플 없이 조인
Dataset<Row> result = users.join(transactions, "user_id");

SparkSession spark = SparkSession.builder()
    .config("spark.shuffle.compress", "true")  // 기본 true
    .config("spark.shuffle.spill.compress", "true")  // 기본 true
    .getOrCreate();

// 빠른 SSD에 셔플 파일 저장
.config("spark.local.dir", "/ssd/spark-local")

// 여러 디스크 분산
.config("spark.local.dir", "/disk1/spark,/disk2/spark,/disk3/spark")

// 최적화 전후 실행 시간 비교
long start = System.currentTimeMillis();
result.count();
long duration = System.currentTimeMillis() - start;
System.out.println("실행 시간: " + duration + "ms");