在 Spark SQL 中，可以通过多种方式使用 SQL 查询 DataFrame，实现灵活的交互式数据分析。

1. 基础 SQL 查询方法

创建临时视图后查询

// 创建 DataFrame
val df = spark.read.option("header", "true").csv("data/employees.csv")

// 注册为临时视图
df.createOrReplaceTempView("employees")

// 使用 SQL 查询
val result = spark.sql("""
  SELECT 
    name, 
    department,
    salary,
    ROUND(salary * 1.1, 2) as new_salary
  FROM employees 
  WHERE salary > 50000 
  ORDER BY salary DESC
""")

result.show()

全局临时视图（跨 Session）

// 创建全局临时视图
df.createGlobalTempView("global_employees")

// 在任何 SparkSession 中查询（需要指定数据库）
spark.sql("SELECT * FROM global_temp.global_employees").show()

// 或者切换到全局数据库
spark.sql("USE global_temp")
spark.sql("SELECT * FROM global_employees").show()

2. 复杂 SQL 查询示例

多表 JOIN 操作

// 准备多个 DataFrame
val employeesDF = spark.read.json("data/employees.json")
val departmentsDF = spark.read.json("data/departments.json")
val salariesDF = spark.read.json("data/salaries.json")

// 注册临时视图
employeesDF.createOrReplaceTempView("employees")
departmentsDF.createOrReplaceTempView("departments") 
salariesDF.createOrReplaceTempView("salaries")

// 复杂 JOIN 查询
val joinResult = spark.sql("""
  SELECT 
    e.employee_id,
    e.name,
    d.department_name,
    s.base_salary,
    s.bonus,
    (s.base_salary + COALESCE(s.bonus, 0)) as total_compensation
  FROM employees e
  INNER JOIN departments d ON e.department_id = d.department_id
  LEFT JOIN salaries s ON e.employee_id = s.employee_id
  WHERE d.department_name IN ('Engineering', 'Sales')
    AND s.base_salary > 60000
  ORDER BY total_compensation DESC
""")

窗口函数和高级分析

val analyticsResult = spark.sql("""
  SELECT 
    department,
    name,
    salary,
    -- 窗口函数：部门内排名
    RANK() OVER (PARTITION BY department ORDER BY salary DESC) as dept_rank,
    -- 聚合函数：部门统计
    AVG(salary) OVER (PARTITION BY department) as avg_dept_salary,
    -- 比较分析
    salary - AVG(salary) OVER (PARTITION BY department) as diff_from_avg,
    -- 累计分布
    CUME_DIST() OVER (PARTITION BY department ORDER BY salary) as percentile
  FROM employees
  WHERE salary IS NOT NULL
  QUALIFY dept_rank <= 5  -- 每个部门前5名
""")

子查询和 CTE（Common Table Expressions）

val cteResult = spark.sql("""
  WITH department_stats AS (
    SELECT 
      department,
      COUNT(*) as employee_count,
      AVG(salary) as avg_salary,
      MAX(salary) as max_salary
    FROM employees
    GROUP BY department
    HAVING COUNT(*) > 10
  ),
  high_earners AS (
    SELECT 
      e.*,
      ds.avg_salary as department_avg
    FROM employees e
    JOIN department_stats ds ON e.department = ds.department
    WHERE e.salary > ds.avg_salary * 1.5
  )
  SELECT 
    department,
    COUNT(*) as high_earner_count,
    ROUND(AVG(salary), 2) as avg_high_earner_salary
  FROM high_earners
  GROUP BY department
  ORDER BY high_earner_count DESC
""")

3. 动态 SQL 查询

参数化查询

// 使用字符串插值
def queryByDepartment(dept: String, minSalary: Double) = {
  spark.sql(s"""
    SELECT name, salary 
    FROM employees 
    WHERE department = '$dept' AND salary > $minSalary
  """)
}

val engineeringHighPaid = queryByDepartment("Engineering", 80000)

// 更安全的参数化方式（防止SQL注入）
def safeQuery(dept: String, minSalary: Double) = {
  val query = """
    SELECT name, salary 
    FROM employees 
    WHERE department = ? AND salary > ?
  """
  spark.sql(query).rdd.map { row =>
    (row.getString(0), row.getDouble(1))
  }.collect()
}

动态表名和列名

// 根据条件动态构建查询
def dynamicQuery(tableName: String, filters: Map[String, Any]) = {
  val whereClause = if (filters.nonEmpty) {
    "WHERE " + filters.map { case (col, value) =>
      value match {
        case s: String => s"$col = '$s'"
        case n: Number => s"$col = $n"
        case _ => s"$col = $value"
      }
    }.mkString(" AND ")
  } else ""
  
  spark.sql(s"SELECT * FROM $tableName $whereClause")
}

// 使用示例
val filteredData = dynamicQuery("employees", Map(
  "department" -> "Engineering", 
  "salary" -> 70000
))

4. SQL 与 DataFrame API 混合使用

链式操作

// 先用 SQL 过滤，再用 DataFrame API 处理
val processedData = spark.sql("""
  SELECT * FROM employees WHERE salary > 50000
""")
  .filter($"department".isin("Engineering", "Data Science"))
  .groupBy("department")
  .agg(
    avg("salary").as("avg_salary"),
    count("*").as("employee_count")
  )
  .orderBy(desc("avg_salary"))

// 或者反过来：先用 DataFrame，再用 SQL
val baseData = employeesDF.select("name", "department", "salary")
baseData.createOrReplaceTempView("filtered_employees")

val finalResult = spark.sql("""
  SELECT department, AVG(salary) as avg_salary
  FROM filtered_employees
  GROUP BY department
  HAVING AVG(salary) > 60000
""")

UDF（用户自定义函数）与 SQL 集成

// 注册 UDF
spark.udf.register("calculate_bonus", (salary: Double, performance: String) => {
  performance match {
    case "excellent" => salary * 0.2
    case "good" => salary * 0.1
    case _ => salary * 0.05
  }
})

// 在 SQL 中使用 UDF
val bonusResult = spark.sql("""
  SELECT 
    name,
    salary,
    performance,
    calculate_bonus(salary, performance) as bonus_amount,
    salary + calculate_bonus(salary, performance) as total_compensation
  FROM employees
  WHERE performance IS NOT NULL
""")

5. 性能优化技巧

缓存常用视图

// 缓存频繁查询的视图
spark.sql("CACHE TABLE employees")

// 或者使用 DataFrame 缓存
val cachedDF = spark.sql("SELECT * FROM employees WHERE salary > 50000").cache()
cachedDF.createOrReplaceTempView("cached_employees")

// 检查缓存状态
spark.sql("CLEAR CACHE")  // 清理缓存

分区和分桶优化

// 如果底层数据是分区的，可以利用分区裁剪
val partitionedQuery = spark.sql("""
  SELECT * FROM partitioned_employees 
  WHERE department = 'Engineering' AND hire_year = 2024
""")

// 启用自适应查询执行
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

执行计划分析

// 查看 SQL 执行计划
val query = spark.sql("SELECT * FROM employees WHERE salary > 50000")
query.explain(true)  // 显示详细执行计划

// 或者直接获取执行计划字符串
val explainString = query.queryExecution.toString()
println(explainString)

6. 实用工具函数

批量 SQL 执行

def executeMultipleQueries(queries: Seq[String]): Unit = {
  queries.foreach { query =>
    try {
      val result = spark.sql(query)
      println(s"执行查询: $query")
      result.show(10, false)
    } catch {
      case e: Exception => 
        println(s"查询执行失败: $query - ${e.getMessage}")
    }
  }
}

// 使用示例
val analysisQueries = Seq(
  "SELECT COUNT(*) as total_employees FROM employees",
  "SELECT department, AVG(salary) as avg_salary FROM employees GROUP BY department",
  "SELECT MAX(salary) as max_salary FROM employees"
)

executeMultipleQueries(analysisQueries)