Spark SQL 实战：从 DataFrame 嵌套到第 5 层，到 SQL 一行流

// 5 层嵌套的 DataFrame——你以为这是 SQL？不，这是噩梦
val result = df1.join(df2, Seq("user_id"), "inner")                // <-- 第 1 层
                .filter($"status" === "active")                     // <-- 第 2 层
                .join(df3, Seq("user_id"), "left")                  // <-- 第 3 层
                .select($"user_id", $"name", $"amount")             // <-- 第 4 层
                .join(df4, Seq("user_id"), "inner")                  // <-- 第 5 层
                .groupBy($"region")
                .agg(sum($"amount").as("total"))

val df = spark.read.parquet("path/to/data.parquet")

// 注册成临时视图
df.createOrReplaceTempView("users") // <-- 现在 users 表可以在 SQL 里用了

// 现在可以用 SQL 查了
spark.sql("SELECT * FROM users WHERE age > 30").show()

// 会话视图：只在当前 spark session 可见
df.createOrReplaceTempView("users")

// 全局视图：跨 session 可见，访问时加 global_temp 前缀
df.createGlobalTempView("users")
spark.sql("SELECT * FROM global_temp.users").show()      // <-- 注意 global_temp. 前缀

// 纯 SQL
val activeUsers = spark.sql("""
  SELECT user_id, name, age
  FROM users
  WHERE status = 'active'
""")

// SQL + DataFrame API 混合：SQL写聚合，DataFrame 写过滤
val summary = spark.sql("""
  SELECT region, AVG(amount) AS avg_amount
  FROM orders
  GROUP BY region
""").filter($"avg_amount" > 1000).orderBy(desc("avg_amount"))   // <-- 混合用法

// ❌ 千万别这么写——SQL 注入风险
val region = "Asia"
spark.sql(s"SELECT * FROM users WHERE region = '$region'")
// 如果 region 是 "'; DROP TABLE users; --" 呢？

// ✅ 安全写法：白名单 + 转义，或直接用 DataFrame API
val filteredDF = df.filter($"region" === region)        // <-- DataFrame API 自动转义

def buildQuery(
    table: String,
    filters: Map[String, Any],
    groupBy: Seq[String],
    orderBy: Seq[String]
): String = {
    val whereClause = if (filters.isEmpty) "1=1" else {
      filters.map { case (k, v) =>
        val escaped = v.toString.replace("'", "''")          // <-- 单引号转义：把 ' 换成 ''
        s"$k = '$escaped'"
      }.mkString(" AND ")
    }
    val selectClause = if (groupBy.isEmpty) "*" else groupBy.mkString(", ")
    val groupByClause = if (groupBy.isEmpty) "" else s"GROUP BY $selectClause"
    val orderByClause = if (orderBy.isEmpty) "" else s"ORDER BY ${orderBy.mkString(", ")}"

    s"""
    SELECT $selectClause, COUNT(*) AS cnt
    FROM $table
    WHERE $whereClause
    $groupByClause
    $orderByClause
    """
}

val sql = buildQuery(
    table = "orders",
    filters = Map("status" -> "active", "region" -> "Asia"),
    groupBy = Seq("region", "product"),
    orderBy = Seq("cnt DESC")
)
spark.sql(sql).show()

// 用白名单限制可用的列名
val allowedColumns = Set("status", "region", "product")

def safeFilter(key: String, value: Any): String = {
  require(allowedColumns.contains(key), s"Column $key is not in whitelist")
  // 单引号转义：把 ' 替换成 ''
  val escaped = value.toString.replace("'", "''")
  s"$key = '$escaped'"
}

df.write
  .mode("overwrite")                                       // <-- 覆盖写模式
  .partitionBy("year", "month")                            // <-- 分区列
  .saveAsTable("warehouse.orders")                          // <-- 写到 spark-warehouse/

// 临时表（session 级）
df.createOrReplaceTempView("orders")

// 持久表（跨 session）
df.write.saveAsTable("orders")                            // <-- 写到 spark-warehouse/orders

// 删除持久表
spark.sql("DROP TABLE IF EXISTS orders")

// 按 year 和 month 分区写入
df.write
  .partitionBy("year", "month")                           // <-- 这两列会成为目录结构
  .mode("overwrite")
  .parquet("path/to/table")

spark.read.parquet("path/to/table").filter($"year" === 2022 && $"month" === 1)
// <-- 只读 year=2022/month=01/ 这一份，省 IO

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://localhost:5432/mydb")
  .option("dbtable", "public.users")
  .option("user", "myuser")
  .option("password", "mypassword")
  .option("numPartitions", "4")        // <-- 并行读 4 个 partition
  .option("partitionColumn", "id")     // <-- 按哪一列分区
  .option("lowerBound", "1")           // <-- id 最小值
  .option("upperBound", "1000000")     // <-- id 最大值
  .load()

df.write
  .format("jdbc")
  .option("url", "jdbc:postgresql://localhost:5432/mydb")
  .option("dbtable", "public.results")
  .option("user", "myuser")
  .option("password", "mypassword")
  .mode("append")          // <-- append 或 overwrite
  .save()

// 并行写：先按 partitionColumn 分区
df.repartition(4, $"partitionColumn")                  // <-- 4 个 partition = 4 个 JDBC 连接
  .write
  .format("jdbc")
  .option("url", "jdbc:postgresql://localhost:5432/mydb")
  .option("dbtable", "public.results")
  .option("user", "myuser")
  .option("password", "mypassword")
  .mode("append")
  .save()

val users = spark.read.parquet("path/to/users").cache()  // <-- 第一次 action 时缓存
users.createOrReplaceTempView("users")                    // <-- 把缓存的 DF 注册成视图

// 多次查询都走缓存
spark.sql("SELECT count(*) FROM users").show()
spark.sql("SELECT region, count(*) FROM users GROUP BY region").show()
spark.sql("SELECT * FROM users WHERE age > 30").show()

import org.apache.spark.storage.StorageLevel

// 等价于 cache()
df.persist(StorageLevel.MEMORY_ONLY)

// 也可以 DISK_ONLY / MEMORY_AND_DISK / MEMORY_ONLY_SER 等
df.persist(StorageLevel.MEMORY_AND_DISK)

df.unpersist()                  // <-- 释放缓存
df.unpersist(blocking = true)   // <-- 同步等缓存释放完