# Spark 버전 확인 (AQE 사용 시 3.0 이상 필요)
spark-submit --version

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

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

// 파티션별 데이터 분포 확인
df.groupBy(spark_partition_id().alias("partition_id"))
    .count()
    .orderBy(col("count").desc())
    .show(20);

// 결과 예시 (스큐 있음):
// +------------+--------+
// |partition_id|   count|
// +------------+--------+
// |          15| 5000000|  ← 비정상적으로 큼!
// |           3|   50000|
// |           7|   48000|
// |           1|   45000|
// ...

// 어떤 키가 핫 키인지 확인
df.groupBy("join_key")
    .count()
    .orderBy(col("count").desc())
    .show(20);

// 결과 예시:
// +--------+--------+
// |join_key|   count|
// +--------+--------+
// |   null | 3000000|  ← null 값이 핫 키!
// | user_1 |  500000|
// | user_2 |   10000|
// ...

SparkSession spark = SparkSession.builder()
    .appName("Skew Fix")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
    .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")
    .getOrCreate();

// 일반 조인 수행 - AQE가 자동으로 스큐 처리
Dataset<Row> result = largeTable.join(smallTable, "key");

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import static org.apache.spark.sql.functions.*;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

int numSaltBuckets = 10;  // 핫 키를 10개 파티션으로 분산

// 1. 큰 테이블에 salt 추가
Dataset<Row> saltedLarge = largeTable.withColumn("salt",
    expr("FLOOR(RAND() * " + numSaltBuckets + ")"))
    .withColumn("salted_key",
        concat(col("join_key"), lit("_"), col("salt")));

// 2. 작은 테이블을 salt 수만큼 복제
List<Row> saltValues = IntStream.range(0, numSaltBuckets)
    .mapToObj(i -> RowFactory.create(i))
    .collect(Collectors.toList());

Dataset<Row> saltDf = spark.createDataFrame(saltValues,
    new StructType().add("salt", DataTypes.IntegerType));

Dataset<Row> replicatedSmall = smallTable.crossJoin(saltDf)
    .withColumn("salted_key",
        concat(col("join_key"), lit("_"), col("salt")));

// 3. salted_key로 조인
Dataset<Row> result = saltedLarge.join(replicatedSmall, "salted_key")
    .drop("salt", "salted_key");

// 핫 키 목록 (사전에 식별)
List<String> hotKeys = Arrays.asList("hot_user_1", "hot_user_2");

Dataset<Row> saltedLarge = largeTable.withColumn("salt",
    when(col("join_key").isin(hotKeys.toArray()),
        expr("FLOOR(RAND() * 10)"))
    .otherwise(lit(0)))
    .withColumn("salted_key",
        concat(col("join_key"), lit("_"), col("salt")));

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

// 작은 테이블을 모든 Executor에 복제
Dataset<Row> result = largeTable.join(
    broadcast(smallTable),  // 셔플 없이 조인
    "key"
);

// 기본 10MB, 필요시 증가
.config("spark.sql.autoBroadcastJoinThreshold", "104857600")  // 100MB

// 1. NULL 분리 처리
Dataset<Row> nonNull = df.filter(col("key").isNotNull());
Dataset<Row> nullRows = df.filter(col("key").isNull());

// 2. non-null만 조인
Dataset<Row> joined = nonNull.join(other, "key");

// 3. null 행 별도 처리 후 합치기
Dataset<Row> result = joined.union(
    nullRows.withColumn("other_col", lit(null))
);

// 처리 후 파티션 분포 재확인
result.groupBy(spark_partition_id())
    .count()
    .agg(
        max("count").alias("max"),
        min("count").alias("min"),
        avg("count").alias("avg")
    )
    .withColumn("skew_ratio", col("max").divide(col("avg")))
    .show();

// skew_ratio가 2~3 이하면 정상