在 Spark SQL 中定义和注册临时视图（Temporary View）

1. 临时视图的基本概念

临时视图是 Spark SQL 中的一种虚拟表，它允许我们将 DataFrame 注册为可以在 SQL 查询中使用的命名表。临时视图不会持久化存储数据，它们仅在当前 SparkSession 的生命周期内存在。

2. 创建临时视图的方法

2.1 createOrReplaceTempView() 方法（推荐）

这是最常用且推荐的方法，因为它会在视图已存在时替换它：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

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

import spark.implicits._

// 创建示例 DataFrame
val employeesDF = Seq(
  (1, "Alice", "Engineering", 75000),
  (2, "Bob", "Marketing", 65000),
  (3, "Charlie", "Engineering", 90000),
  (4, "David", "Sales", 55000)
).toDF("id", "name", "department", "salary")

// 创建或替换临时视图
employeesDF.createOrReplaceTempView("employees")

// 现在可以使用 SQL 查询这个视图
val result = spark.sql("SELECT * FROM employees WHERE salary > 70000")
result.show()

2.2 createTempView() 方法

这种方法只创建新视图，如果视图已存在会抛出异常：

// 创建新的临时视图（如果已存在会报错）
try {
  employeesDF.createTempView("employees_new")
  println("Temporary view created successfully")
} catch {
  case e: Exception => 
    println(s"Failed to create temporary view: ${e.getMessage}")
}

// 查询新建的视图
val newResult = spark.sql("SELECT name, department FROM employees_new")
newResult.show()

2.3 createOrReplaceGlobalTempView() 方法

创建全局临时视图，在所有 SparkSession 中都可见：

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

// 查询全局临时视图需要加上 global_temp 前缀
val globalResult = spark.sql("SELECT * FROM global_temp.global_employees")
globalResult.show()

// 在其他 SparkSession 中也可以访问（如果有多个 Session）

2.4 createGlobalTempView() 方法

创建新的全局临时视图，如果已存在会抛出异常：

// 创建新的全局临时视图
try {
  employeesDF.createGlobalTempView("new_global_employees")
  println("Global temporary view created successfully")
} catch {
  case e: Exception =>
    println(s"Failed to create global temporary view: ${e.getMessage}")
}

3. 临时视图的详细特性

3.1 生命周期管理

// 演示临时视图的生命周期
object TempViewLifecycleDemo {
  
  def demonstrateLifecycle(): Unit = {
    val spark = SparkSession.builder()
      .appName("Lifecycle Demo")
      .master("local[*]")
      .getOrCreate()
    
    import spark.implicits._
    
    val df = Seq(("Test", 100)).toDF("name", "value")
    
    // 创建临时视图
    df.createOrReplaceTempView("temp_table")
    
    // 验证视图存在
    val catalog = spark.catalog
    println(s"View exists: ${catalog.tableExists("temp_table")}")
    
    // 列出所有临时视图
    println("All tables in catalog:")
    catalog.listTables().show()
    
    // 删除临时视图
    catalog.dropTempView("temp_table")
    println(s"After drop - View exists: ${catalog.tableExists("temp_table")}")
    
    // 删除全局临时视图
    catalog.dropGlobalTempView("global_temp_table")
  }
}

3.2 作用域区别

// 本地临时视图 vs 全局临时视图
object ScopeComparison {
  
  def compareScopes(): Unit = {
    val spark1 = SparkSession.builder()
      .appName("Session 1")
      .master("local[*]")
      .getOrCreate()
      
    val spark2 = SparkSession.builder()
      .appName("Session 2")
      .master("local[*]")
      .getOrCreate()
    
    import spark1.implicits._
    import spark2.implicits._
    
    val df1 = Seq(("A", 1)).toDF("letter", "number")
    val df2 = Seq(("B", 2)).toDF("letter", "number")
    
    // 本地临时视图 - 只在当前 Session 可见
    df1.createOrReplaceTempView("local_view")
    
    // 全局临时视图 - 所有 Session 可见
    df2.createOrReplaceGlobalTempView("global_view")
    
    // Session 1 可以访问本地和全局视图
    println("Session 1 local view:")
    spark1.sql("SELECT * FROM local_view").show()
    
    println("Session 1 global view:")
    spark1.sql("SELECT * FROM global_temp.global_view").show()
    
    // Session 2 不能访问 Session 1 的本地视图
    try {
      spark2.sql("SELECT * FROM local_view").show()
    } catch {
      case e: Exception => println(s"Cannot access local view: ${e.getMessage}")
    }
    
    // 但可以访问全局视图
    println("Session 2 global view:")
    spark2.sql("SELECT * FROM global_temp.global_view").show()
  }
}

4. 实际应用示例

4.1 多步骤数据分析流程

object MultiStepAnalysis {
  
  def performAnalysis(): Unit = {
    val spark = SparkSession.builder()
      .appName("Multi-step Analysis")
      .master("local[*]")
      .getOrCreate()
    
    import spark.implicits._
    
    // 原始数据
    val salesData = Seq(
      ("2023-01-01", "Product A", 100, 25.50),
      ("2023-01-02", "Product B", 200, 15.75),
      ("2023-01-03", "Product A", 150, 25.50),
      ("2023-01-04", "Product C", 75, 35.00)
    ).toDF("date", "product", "quantity", "price")
    
    // 步骤1: 创建基础销售数据视图
    salesData.createOrReplaceTempView("raw_sales")
    
    // 步骤2: 创建每日汇总视图
    spark.sql("""
      CREATE OR REPLACE TEMP VIEW daily_summary AS
      SELECT 
        date,
        product,
        quantity,
        price,
        quantity * price as revenue
      FROM raw_sales
    """)
    
    // 步骤3: 创建产品汇总视图
    spark.sql("""
      CREATE OR REPLACE TEMP VIEW product_summary AS
      SELECT 
        product,
        SUM(quantity) as total_quantity,
        SUM(revenue) as total_revenue,
        AVG(price) as avg_price,
        COUNT(*) as transaction_count
      FROM daily_summary
      GROUP BY product
    """)
    
    // 最终查询：获取产品表现报告
    val finalReport = spark.sql("""
      SELECT 
        product,
        total_quantity,
        ROUND(total_revenue, 2) as revenue,
        ROUND(avg_price, 2) as average_price,
        transaction_count,
        RANK() OVER (ORDER BY total_revenue DESC) as revenue_rank
      FROM product_summary
      ORDER BY total_revenue DESC
    """)
    
    finalReport.show()
  }
}

4.2 复杂业务逻辑实现

object ComplexBusinessLogic {
  
  def implementBusinessRules(): Unit = {
    val spark = SparkSession.builder()
      .appName("Business Logic Implementation")
      .master("local[*]")
      .getOrCreate()
    
    import spark.implicits._
    
    // 客户数据
    val customers = Seq(
      (1, "Alice", "Premium", 10000),
      (2, "Bob", "Regular", 5000),
      (3, "Charlie", "VIP", 25000),
      (4, "David", "Regular", 3000)
    ).toDF("customer_id", "name", "tier", "credit_limit")
    
    // 订单数据
    val orders = Seq(
      (101, 1, 1500.00, "2023-01-15"),
      (102, 2, 800.00, "2023-01-16"),
      (103, 1, 2500.00, "2023-01-17"),
      (104, 3, 5000.00, "2023-01-18")
    ).toDF("order_id", "customer_id", "amount", "order_date")
    
    // 注册视图
    customers.createOrReplaceTempView("customers")
    orders.createOrReplaceTempView("orders")
    
    // 创建客户订单汇总视图
    spark.sql("""
      CREATE OR REPLACE TEMP VIEW customer_order_summary AS
      SELECT 
        c.customer_id,
        c.name,
        c.tier,
        c.credit_limit,
        COUNT(o.order_id) as order_count,
        SUM(o.amount) as total_spent,
        MAX(o.order_date) as last_order_date,
        AVG(o.amount) as avg_order_value
      FROM customers c
      LEFT JOIN orders o ON c.customer_id = o.customer_id
      GROUP BY c.customer_id, c.name, c.tier, c.credit_limit
    """)
    
    // 创建风险评估视图
    spark.sql("""
      CREATE OR REPLACE TEMP VIEW risk_assessment AS
      SELECT 
        *,
        CASE 
          WHEN total_spent > credit_limit * 0.8 THEN 'HIGH_RISK'
          WHEN total_spent > credit_limit * 0.5 THEN 'MEDIUM_RISK'
          ELSE 'LOW_RISK'
        END as risk_level,
        ROUND((total_spent / credit_limit) * 100, 2) as utilization_rate
      FROM customer_order_summary
    """)
    
    // 最终业务报告
    val businessReport = spark.sql("""
      SELECT 
        name,
        tier,
        credit_limit,
        COALESCE(total_spent, 0) as total_spent,
        order_count,
        risk_level,
        utilization_rate,
        CASE 
          WHEN tier = 'VIP' AND risk_level = 'LOW_RISK' THEN 'PROMOTION_ELIGIBLE'
          WHEN tier = 'Premium' AND utilization_rate > 50 THEN 'UPGRADE_CANDIDATE'
          ELSE 'REGULAR_MONITORING'
        END as action_recommendation
      FROM risk_assessment
      ORDER BY total_spent DESC NULLS LAST
    """)
    
    businessReport.show()
  }
}

5. 性能考虑和最佳实践

5.1 缓存策略

// 对于频繁查询的临时视图，考虑缓存底层 DataFrame
val frequentlyUsedDF = spark.sql("SELECT * FROM complex_derived_view")
frequentlyUsedDF.cache()
frequentlyUsedDF.count() // 触发缓存

// 或者直接缓存视图的结果
spark.sql("CACHE TABLE frequently_used_view AS SELECT * FROM complex_query")

5.2 视图管理

object ViewManagement {
  
  def manageViews(spark: SparkSession): Unit = {
    val catalog = spark.catalog
    
    // 列出所有视图
    println("Available views:")
    catalog.listTables()
      .filter($"tableType" === "TEMPORARY")
      .select("name", "description")
      .show(false)
    
    // 检查特定视图是否存在
    val viewName = "important_analysis_view"
    if (catalog.tableExists(viewName)) {
      println(s"View $viewName exists")
    } else {
      println(s"View $viewName does not exist")
    }
    
    // 清理不再需要的视图
    val viewsToDrop = Seq("temp_view_1", "temp_view_2", "old_analysis")
    viewsToDrop.foreach { view =>
      if (catalog.tableExists(view)) {
        catalog.dropTempView(view)
        println(s"Dropped view: $view")
      }
    }
  }
}

5.3 错误处理

object ErrorHandling {
  
  def safeViewCreation(df: org.apache.spark.sql.DataFrame, viewName: String): Boolean = {
    try {
      df.createOrReplaceTempView(viewName)
      println(s"Successfully created/updated view: $viewName")
      true
    } catch {
      case e: Exception =>
        println(s"Failed to create view $viewName: ${e.getMessage}")
        false
    }
  }
  
  def safeSqlExecution(spark: SparkSession, query: String): Option[org.apache.spark.sql.DataFrame] = {
    try {
      val result = spark.sql(query)
      Some(result)
    } catch {
      case e: Exception =>
        println(s"SQL execution failed: ${e.getMessage}")
        None
    }
  }
}

6. 高级用法

6.1 动态视图创建

object DynamicViewCreation {
  
  def createViewsFromMultipleSources(): Unit = {
    val spark = SparkSession.builder()
      .appName("Dynamic Views")
      .master("local[*]")
      .getOrCreate()
    
    // 模拟多个数据源
    val sources = Map(
      "sales" -> Seq(("Product A", 100), ("Product B", 200)).toDF("product", "quantity"),
      "inventory" -> Seq(("Product A", 50), ("Product B", 150)).toDF("product", "stock"),
      "prices" -> Seq(("Product A", 25.50), ("Product B", 15.75)).toDF("product", "price")
    )
    
    // 动态创建视图
    sources.foreach { case (sourceName, df) =>
      val viewName = s"${sourceName}_view"
      df.createOrReplaceTempView(viewName)
      println(s"Created view: $viewName")
    }
    
    // 跨源分析查询
    val crossSourceAnalysis = spark.sql("""
      SELECT 
        s.product,
        s.quantity,
        i.stock,
        p.price,
        s.quantity * p.price as potential_revenue,
        i.stock - s.quantity as stock_balance
      FROM sales_view s
      JOIN inventory_view i ON s.product = i.product
      JOIN prices_view p ON s.product = p.product
      ORDER BY potential_revenue DESC
    """)
    
    crossSourceAnalysis.show()
  }
}

通过以上详细介绍，您现在应该完全掌握了在 Spark SQL 中定义和注册临时视图的各种方法、特性和最佳实践。临时视图是连接 DataFrame API 和 SQL 查询的重要桥梁，正确使用它们可以让您的数据分析工作更加高效和灵活。