在 Spark SQL 中定义和注册临时视图有多种方式，临时视图的生命周期与 SparkSession 绑定。

1. 基础临时视图注册

从 DataFrame 创建临时视图

// 创建示例 DataFrame
val employeesDF = Seq(
  ("Alice", "Engineering", 75000),
  ("Bob", "Sales", 65000),
  ("Charlie", "Engineering", 80000)
).toDF("name", "department", "salary")

// 注册为临时视图（方法一：createOrReplaceTempView）
employeesDF.createOrReplaceTempView("employees")

// 注册为临时视图（方法二：createTempView - 如果已存在会报错）
employeesDF.createTempView("employees_temp")

// 立即使用 SQL 查询
spark.sql("SELECT * FROM employees WHERE salary > 70000").show()

全局临时视图（跨 Session 共享）

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

// 查询全局临时视图（需要指定数据库）
spark.sql("SELECT * FROM global_temp.global_employees").show()

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

// 在其他 SparkSession 中也可以访问
val newSpark = SparkSession.builder().getOrCreate()
newSpark.sql("SELECT * FROM global_temp.global_employees").show()

2. 高级视图注册选项

带选项的视图创建

// 创建视图时指定属性
employeesDF.createOrReplaceTempView("employees_view")

// 通过 SQL 设置视图属性
spark.sql("""
  CREATE OR REPLACE TEMPORARY VIEW employees_with_props
  COMMENT '员工信息视图'
  TBLPROPERTIES ('creator'='data_team', 'created_date'='2024-01-01')
  AS SELECT * FROM employees
""")

基于查询结果的视图

// 直接通过 SQL 创建视图（无需先有 DataFrame）
spark.sql("""
  CREATE OR REPLACE TEMPORARY VIEW high_paid_employees AS
  SELECT 
    name,
    department,
    salary,
    ROUND(salary * 1.1, 2) as expected_salary
  FROM employees 
  WHERE salary > 70000
  ORDER BY salary DESC
""")

// 查询新创建的视图
spark.sql("SELECT * FROM high_paid_employees").show()

3. 视图管理操作

检查视图是否存在

// 方法一：使用 catalog API
val catalog = spark.catalog
if (catalog.tableExists("employees")) {
  println("employees 视图存在")
}

// 方法二：尝试查询（捕获异常）
try {
  spark.sql("SELECT 1 FROM employees LIMIT 1")
  println("视图存在且可访问")
} catch {
  case e: Exception => println(s"视图不存在或不可访问: ${e.getMessage}")
}

列出所有临时视图

// 列出当前数据库的临时视图
spark.sql("SHOW TABLES").show()

// 使用 catalog API 获取详细信息
val tables = catalog.listTables()
tables.filter(_.tableType == "TEMPORARY").show(false)

// 查看视图结构
spark.sql("DESCRIBE employees").show()

删除临时视图

// 方法一：使用 SQL DROP
spark.sql("DROP VIEW IF EXISTS employees")

// 方法二：使用 catalog API（间接方式）
// 临时视图会在 SparkSession 关闭时自动清理，通常不需要手动删除

// 删除全局临时视图
spark.sql("DROP VIEW IF EXISTS global_temp.global_employees")

4. 实际应用场景

场景1：数据预处理管道

// 原始数据
val rawDataDF = spark.read.json("data/raw_logs.json")

// 第一步：清洗数据并创建视图
rawDataDF
  .filter($"timestamp".isNotNull)
  .na.fill("unknown", Seq("user_id"))
  .createOrReplaceTempView("cleaned_logs")

// 第二步：聚合分析
spark.sql("""
  CREATE TEMPORARY VIEW daily_stats AS
  SELECT 
    DATE(timestamp) as date,
    COUNT(*) as total_events,
    COUNT(DISTINCT user_id) as unique_users
  FROM cleaned_logs
  GROUP BY DATE(timestamp)
""")

// 第三步：最终查询
val finalResult = spark.sql("""
  SELECT * FROM daily_stats WHERE unique_users > 1000
""")

场景2：多步骤数据分析

// 准备基础数据
val salesDF = spark.read.parquet("data/sales.parquet")
val productsDF = spark.read.parquet("data/products.parquet")

// 注册基础视图
salesDF.createOrReplaceTempView("sales")
productsDF.createOrReplaceTempView("products")

// 创建中间分析视图
spark.sql("""
  CREATE TEMPORARY VIEW product_sales AS
  SELECT 
    p.product_name,
    p.category,
    s.sale_date,
    s.quantity,
    s.amount
  FROM sales s
  JOIN products p ON s.product_id = p.product_id
""")

// 创建汇总视图
spark.sql("""
  CREATE TEMPORARY VIEW category_summary AS
  SELECT 
    category,
    COUNT(*) as total_sales,
    SUM(amount) as total_revenue,
    AVG(amount) as avg_sale_amount
  FROM product_sales
  GROUP BY category
""")

// 最终业务查询
val businessReport = spark.sql("""
  SELECT 
    category,
    total_sales,
    total_revenue,
    ROUND(avg_sale_amount, 2) as avg_sale_amount,
    ROUND(total_revenue / total_sales, 2) as revenue_per_sale
  FROM category_summary
  ORDER BY total_revenue DESC
""")

场景3：动态视图创建

// 根据配置动态创建视图
def createDynamicView(viewName: String, sourceTable: String, filters: Map[String, Any]) = {
  val whereClause = if (filters.nonEmpty) {
    "WHERE " + filters.map { case (col, value) =>
      s"$col = '$value'"
    }.mkString(" AND ")
  } else ""
  
  val createViewSQL = s"""
    CREATE OR REPLACE TEMPORARY VIEW $viewName AS
    SELECT * FROM $sourceTable $whereClause
  """
  
  spark.sql(createViewSQL)
  println(s"已创建视图: $viewName")
}

// 使用示例
createDynamicView("engineering_sales", "product_sales", 
  Map("category" -> "Engineering", "sale_date" -> "2024-01-15"))

5. 性能优化和最佳实践

视图缓存策略

// 缓存频繁使用的视图对应的 DataFrame
val cachedDF = spark.sql("SELECT * FROM large_table WHERE condition = true").cache()
cachedDF.createOrReplaceTempView("cached_large_table")

// 或者直接缓存视图
spark.sql("CACHE TABLE cached_large_table")

分区视图优化

// 如果底层数据是分区的，创建分区视图可以提高查询性能
spark.sql("""
  CREATE TEMPORARY VIEW partitioned_sales AS
  SELECT * FROM sales 
  WHERE sale_date >= '2024-01-01' 
    AND sale_date <= '2024-01-31'
""")

// 启用分区裁剪
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

视图生命周期管理

// 在长时间运行的作业中管理视图生命周期
class ViewManager(spark: SparkSession) {
  private var registeredViews = scala.collection.mutable.Set[String]()
  
  def registerView(df: DataFrame, viewName: String): Unit = {
    df.createOrReplaceTempView(viewName)
    registeredViews.add(viewName)
  }
  
  def cleanup(): Unit = {
    registeredViews.foreach { viewName =>
      try {
        spark.sql(s"DROP VIEW IF EXISTS $viewName")
      } catch {
        case e: Exception => println(s"清理视图 $viewName 失败: ${e.getMessage}")
      }
    }
    registeredViews.clear()
  }
}

// 使用示例
val viewManager = new ViewManager(spark)
viewManager.registerView(employeesDF, "managed_employees")

// 作业完成后清理
// viewManager.cleanup()

临时视图是 Spark SQL 中非常重要的概念，它们提供了一种轻量级的方式来组织和重用数据转换逻辑，同时保持了代码的清晰性和可维护性。