Python使用Spark编写单元测试时如何为临时视图指定数据库前缀

Spark 的临时视图（包括会话级临时视图、全局临时视图）本身有独立的命名空间，不支持指定用户自定义的数据库/ Schema 前缀，该报错由Spark临时视图的原生特性导致：

• 会话级临时视图仅当前Spark会话可见，没有归属的自定义库
• 全局临时视图归属系统固定内置库global_temp，不支持修改前缀

不需要修改原有业务代码的查询逻辑，两种方案可选： #

方案1：创建对应库并写入测试表（最贴合真实业务场景，推荐） #

直接创建union库，把测试数据写入该库下的items_vw表，完全匹配业务的库表结构，单元测试逻辑和线上逻辑完全一致：

from pyspark.sql import SparkSession, Row
from datetime import datetime

columns = ["company_id", "item_id", "ingest_date", "attachable_id"]
data = [
    (100, 1, datetime(2021, 1, 1, 0, 0, 0, 0), 900),
    (100, 2, datetime(2021, 1, 1, 0, 3, 0, 0), 903),
    (100, 2, datetime(2021, 1, 1, 0, 1, 0, 0), 902),
    (200, 1, datetime(2021, 1, 1, 0, 1, 0, 0), 900)
]
rowData = map(lambda x: Row(*x), data)
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
df = spark.createDataFrame(rowData, columns)

# 新增逻辑：创建union库，写入测试表
spark.sql("CREATE SCHEMA IF NOT EXISTS union")
df.write.mode("overwrite").saveAsTable("union.items_vw")

select_query = "SELECT count(*) from union.items_vw"
batch_df = spark.sql(select_query)
print(batch_df.show())

方案2：切换默认库+临时视图（无磁盘IO，速度更快） #

如果不想落盘测试数据，只想用内存级的临时视图，可以先把当前会话的默认库切换为union，再创建不带前缀的临时视图，此时查询union.items_vw会自动命中创建的临时视图：

# ... 前面创建df的逻辑不变 ...
spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
df = spark.createDataFrame(rowData, columns)

# 新增逻辑：创建union库并设置为当前会话默认库
spark.sql("CREATE SCHEMA IF NOT EXISTS union")
spark.sql("USE union")
# 创建不带前缀的临时视图
df.createOrReplaceTempView("items_vw")

select_query = "SELECT count(*) from union.items_vw"
batch_df = spark.sql(select_query)
print(batch_df.show())

内容的提问来源于stack exchange，提问作者Raju