在 Spark SQL 中使用 SQL 查询 DataFrame

1. 基础概念与原理

在 Spark SQL 中，可以通过两种主要方式操作结构化数据：

1. DataFrame API - 使用编程接口
2. SQL 查询 - 使用标准 SQL 语法

这两种方式可以无缝互换使用，底层都经过相同的 Catalyst 优化器处理。

2. 创建临时视图

要使用 SQL 查询 DataFrame，首先需要将其注册为临时视图：

2.1 创建临时视图的方法

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("SQL Query Example")
  .master("local[*]")
  .getOrCreate()

import spark.implicits._

// 创建示例 DataFrame
val employees = Seq(
  (1, "Alice", 25, "Engineering", 75000),
  (2, "Bob", 30, "Marketing", 65000),
  (3, "Charlie", 35, "Engineering", 90000),
  (4, "David", 28, "Sales", 55000),
  (5, "Eve", 32, "Marketing", 70000)
).toDF("id", "name", "age", "department", "salary")

// 方法1: createOrReplaceTempView（推荐）
employees.createOrReplaceTempView("employees_view")

// 方法2: createTempView
employees.createTempView("employees_temp")

// 方法3: createGlobalTempView（全局临时视图）
employees.createGlobalTempView("global_employees")

2.2 临时视图的作用域和生命周期

// 本地临时视图 - 只在当前 SparkSession 中可见
employees.createOrReplaceTempView("local_view")
spark.sql("SELECT * FROM local_view").show()

// 全局临时视图 - 在所有 SparkSession 中可见
employees.createGlobalTempView("global_view")
spark.sql("SELECT * FROM global_temp.global_view").show()

// 检查视图是否存在
val catalog = spark.catalog
println(s"Local view exists: ${catalog.tableExists("local_view")}")
println(s"Global view exists: ${catalog.tableExists("global_temp.global_view")}")

// 删除临时视图
spark.catalog.dropTempView("local_view")
spark.catalog.dropGlobalTempView("global_view")

3. 基本 SQL 查询操作

3.1 简单查询

// 全表查询
val allEmployees = spark.sql("SELECT * FROM employees_view")
allEmployees.show()

// 选择特定列
val selectedColumns = spark.sql("""
  SELECT name, department, salary 
  FROM employees_view
""")
selectedColumns.show()

// 添加计算列
val calculatedFields = spark.sql("""
  SELECT 
    name,
    department,
    salary,
    salary * 12 as annual_salary,
    ROUND(salary * 0.1, 2) as bonus
  FROM employees_view
""")
calculatedFields.show()

3.2 过滤查询

// WHERE 条件过滤
val filteredResults = spark.sql("""
  SELECT * 
  FROM employees_view 
  WHERE salary > 70000
""")
filteredResults.show()

// 多条件过滤
val complexFilter = spark.sql("""
  SELECT name, department, salary
  FROM employees_view
  WHERE age BETWEEN 25 AND 35
    AND department IN ('Engineering', 'Marketing')
    AND salary >= 65000
  ORDER BY salary DESC
""")
complexFilter.show()

3.3 聚合查询

// 基本聚合函数
val aggregationQuery = spark.sql("""
  SELECT 
    department,
    COUNT(*) as employee_count,
    AVG(salary) as avg_salary,
    MIN(salary) as min_salary,
    MAX(salary) as max_salary,
    SUM(salary) as total_salary
  FROM employees_view
  GROUP BY department
""")
aggregationQuery.show()

// 使用 HAVING 子句
val havingQuery = spark.sql("""
  SELECT 
    department,
    AVG(salary) as avg_salary,
    COUNT(*) as employee_count
  FROM employees_view
  GROUP BY department
  HAVING COUNT(*) >= 2
  ORDER BY avg_salary DESC
""")
havingQuery.show()

4. 高级 SQL 查询功能

4.1 连接查询

// 创建另一个示例 DataFrame
val departments = Seq(
  (1, "Engineering", "Building A"),
  (2, "Marketing", "Building B"),
  (3, "Sales", "Building C"),
  (4, "HR", "Building D")
).toDF("dept_id", "dept_name", "location")

departments.createOrReplaceTempView("departments_view")

// 内连接
val innerJoin = spark.sql("""
  SELECT 
    e.name,
    e.salary,
    d.dept_name,
    d.location
  FROM employees_view e
  INNER JOIN departments_view d ON e.department = d.dept_name
""")
innerJoin.show()

// 左外连接
val leftJoin = spark.sql("""
  SELECT 
    e.name,
    e.department,
    d.location,
    CASE 
      WHEN d.location IS NULL THEN 'Unknown Department'
      ELSE d.location
    END as dept_location
  FROM employees_view e
  LEFT JOIN departments_view d ON e.department = d.dept_name
""")
leftJoin.show()

// 自连接示例
employees.createOrReplaceTempView("emp_self")
val selfJoin = spark.sql("""
  SELECT 
    e1.name as employee_name,
    e1.salary as employee_salary,
    e2.name as manager_name,
    e2.salary as manager_salary
  FROM emp_self e1
  LEFT JOIN emp_self e2 ON e1.id = e2.id - 1
  WHERE e1.id < 5
""")
selfJoin.show()

4.2 子查询

// 标量子查询
val scalarSubquery = spark.sql("""
  SELECT 
    name,
    salary,
    (SELECT AVG(salary) FROM employees_view) as company_avg,
    salary - (SELECT AVG(salary) FROM employees_view) as diff_from_avg
  FROM employees_view
  WHERE salary > (SELECT AVG(salary) FROM employees_view)
""")
scalarSubquery.show()

// 行子查询
val rowSubquery = spark.sql("""
  SELECT *
  FROM employees_view
  WHERE (department, salary) IN (
    SELECT department, MAX(salary)
    FROM employees_view
    GROUP BY department
  )
""")
rowSubquery.show()

// 相关子查询
val correlatedSubquery = spark.sql("""
  SELECT 
    name,
    department,
    salary
  FROM employees_view e1
  WHERE salary > (
    SELECT AVG(salary)
    FROM employees_view e2
    WHERE e2.department = e1.department
  )
""")
correlatedSubquery.show()

4.3 窗口函数

// 排名窗口函数
val rankingWindow = spark.sql("""
  SELECT 
    name,
    department,
    salary,
    ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) as dept_rank,
    RANK() OVER (ORDER BY salary DESC) as overall_rank,
    DENSE_RANK() OVER (ORDER BY salary DESC) as dense_rank
  FROM employees_view
""")
rankingWindow.show()

// 分析窗口函数
val analyticalWindow = spark.sql("""
  SELECT 
    name,
    department,
    salary,
    AVG(salary) OVER (PARTITION BY department) as dept_avg_salary,
    SUM(salary) OVER (PARTITION BY department ORDER BY salary) as cumulative_dept_salary,
    LAG(salary, 1) OVER (PARTITION BY department ORDER BY age) as prev_salary,
    LEAD(age, 1) OVER (PARTITION BY department ORDER BY age) as next_age
  FROM employees_view
""")
analyticalWindow.show()

// NTILE 函数
val ntileWindow = spark.sql("""
  SELECT 
    name,
    salary,
    NTILE(3) OVER (ORDER BY salary) as salary_quartile
  FROM employees_view
""")
ntileWindow.show()

4.4 集合操作

// UNION 操作
val highEarners = spark.sql("""
  SELECT name, salary, 'High Earner' as category
  FROM employees_view
  WHERE salary >= 75000
""")

val lowEarners = spark.sql("""
  SELECT name, salary, 'Low Earner' as category
  FROM employees_view
  WHERE salary < 75000
""")

highEarners.createOrReplaceTempView("high_earners")
lowEarners.createOrReplaceTempView("low_earners")

val unionResult = spark.sql("""
  SELECT * FROM high_earners
  UNION ALL
  SELECT * FROM low_earners
  ORDER BY salary DESC
""")
unionResult.show()

// INTERSECT 和 EXCEPT 示例
val intersectResult = spark.sql("""
  SELECT department FROM employees_view WHERE salary > 70000
  INTERSECT
  SELECT department FROM employees_view WHERE age < 35
""")
intersectResult.show()

5. 复杂查询示例

5.1 CTE (Common Table Expressions)

val cteQuery = spark.sql("""
  WITH dept_stats AS (
    SELECT 
      department,
      COUNT(*) as emp_count,
      AVG(salary) as avg_salary
    FROM employees_view
    GROUP BY department
  ),
  high_performers AS (
    SELECT 
      e.*,
      ds.avg_salary as dept_avg_salary
    FROM employees_view e
    JOIN dept_stats ds ON e.department = ds.department
    WHERE e.salary > ds.avg_salary
  )
  SELECT 
    name,
    department,
    salary,
    ROUND(dept_avg_salary, 2) as dept_average,
    ROUND(salary - dept_avg_salary, 2) as above_avg_amount
  FROM high_performers
  ORDER BY salary DESC
""")
cteQuery.show()

5.2 复杂分析查询

val complexAnalysis = spark.sql("""
  SELECT 
    department,
    name,
    salary,
    age,
    -- 部门内排名
    ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) as dept_salary_rank,
    -- 公司内排名
    DENSE_RANK() OVER (ORDER BY salary DESC) as company_salary_rank,
    -- 百分位排名
    PERCENT_RANK() OVER (ORDER BY salary) as salary_percentile,
    -- 部门平均薪资
    AVG(salary) OVER (PARTITION BY department) as dept_avg_salary,
    -- 薪资差异
    salary - AVG(salary) OVER (PARTITION BY department) as diff_from_dept_avg,
    -- 累计分布
    CUME_DIST() OVER (ORDER BY salary) as cumulative_distribution
  FROM employees_view
  QUALIFY ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) <= 2
  ORDER BY department, salary DESC
""")
complexAnalysis.show()

6. 性能优化技巧

6.1 查询计划分析

// 查看逻辑计划和物理计划
val df = spark.sql("SELECT * FROM employees_view WHERE salary > 70000")

println("=== Logical Plan ===")
df.explain("logical")

println("\n=== Physical Plan ===")
df.explain("physical")

println("\n=== Extended Plan ===")
df.explain("extended")

6.2 优化提示(Hints)

// 广播提示
val broadcastHintQuery = spark.sql("""
  SELECT /*+ BROADCAST(departments_view) */ 
    e.name,
    d.location
  FROM employees_view e
  JOIN departments_view d ON e.department = d.dept_name
""")

// 分区提示
val partitionHintQuery = spark.sql("""
  SELECT /*+ COALESCE(5) */
    department,
    AVG(salary) as avg_salary
  FROM employees_view
  GROUP BY department
""")

6.3 缓存策略

// 缓存频繁使用的 DataFrame
val cachedDF = spark.sql("SELECT * FROM employees_view").cache()
cachedDF.count() // 触发缓存

// 使用后清理缓存
// cachedDF.unpersist()

7. 实际应用场景

7.1 数据探索分析

// 数据质量检查
val dataQualityCheck = spark.sql("""
  SELECT 
    COUNT(*) as total_records,
    COUNT(DISTINCT id) as unique_ids,
    COUNT(CASE WHEN name IS NULL THEN 1 END) as null_names,
    COUNT(CASE WHEN salary <= 0 THEN 1 END) as invalid_salaries,
    MIN(age) as min_age,
    MAX(age) as max_age
  FROM employees_view
""")
dataQualityCheck.show()

// 分布统计
val distributionStats = spark.sql("""
  SELECT 
    department,
    COUNT(*) as count,
    ROUND(AVG(salary), 2) as avg_salary,
    ROUND(STDDEV(salary), 2) as std_dev_salary,
    MIN(salary) as min_salary,
    MAX(salary) as max_salary
  FROM employees_view
  GROUP BY department
  ORDER BY avg_salary DESC
""")
distributionStats.show()

7.2 业务指标计算

// KPI 计算示例
val kpiMetrics = spark.sql("""
  SELECT 
    CURRENT_DATE() as report_date,
    COUNT(*) as total_employees,
    ROUND(AVG(salary), 2) as average_salary,
    ROUND(
      (COUNT(CASE WHEN salary > 75000 THEN 1 END) * 100.0 / COUNT(*)), 2
    ) as high_earner_percentage,
    FIRST(department ORDER BY salary DESC) as highest_paid_department,
    SUM(salary) as total_payroll
  FROM employees_view
""")
kpiMetrics.show()

通过以上详细介绍，您可以看到在 Spark SQL 中使用 SQL 查询 DataFrame 是非常强大和灵活的。这种方式不仅提供了熟悉的 SQL 语法，还充分利用了 Spark 的分布式计算能力和优化器的优势，使得复杂的分析查询变得简单高效。