PySpark无法找到spark-warehouse内的bronze数据库，求解决方案

问题场景 #

现有名为bronze的数据库，运行下述代码时，执行df_new.write.mode("overwrite").format("delta").saveAsTable(f"bronze.{TABLE_NAME}")语句抛出错误：

pyspark.sql.utils.AnalysisException: Database 'bronze' not found

完整代码如下：

import findspark
findspark.init()

import delta
import os
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType, FloatType
from pyspark.sql import SparkSession, window
from pyspark.sql import functions as F
from os.path import abspath

def upsertToDelta(df, batchId):
  '''
  In order to guarantee there aren't any duplicated matches, a Window is used to filter matches based on its GameId and UpdatedUtc.
  The GameId is used as a group by and UpdatedUtc is used as an order by.
  If it's found a duplicated match, the duplicate will be not be saved.
  '''
  windowSpec = window.Window.partitionBy("GameId").orderBy("UpdatedUtc") # .orderBy(1)
  df_new = df.withColumn("row_number", F.row_number().over(windowSpec)).filter("row_number = 1")

  ( bronzeDeltaTable.alias("bronze")
                    .merge(df_new.alias("raw"), "bronze.GameId = raw.GameId")
                    .whenMatchedUpdateAll()
                    .whenNotMatchedInsertAll()
                    .execute()
  )

warehouse_location = abspath('spark-warehouse')

builder = SparkSession.builder \
    .master('local[*]') \
    .config("spark.sql.warehouse.dir", warehouse_location) \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = delta.configure_spark_with_delta_pip(builder) \
        .getOrCreate()

leaderboards_schema = StructType([
    StructField("PlayerId", IntegerType(), False),
    StructField("TeamId", IntegerType(), False),
    StructField("Name", StringType(), False),
    StructField("MatchName", StringType(), False),
    StructField("Team", StringType(), False),
    StructField("IsClosed", BooleanType(), False),
    StructField("GameId", IntegerType(), False),
    StructField("OpponentId", IntegerType(), False),
    StructField("Opponent", StringType(), False),
    StructField("Day", StringType(), True),
    StructField("DateTime", StringType(), True),
    StructField("Updated", StringType(), True),
    StructField("UpdatedUtc", StringType(), True),
    StructField("Games", StringType(), True),
    StructField("Maps", FloatType(), True),
    StructField("FantasyPoints", FloatType(), True),
    StructField("Kills", FloatType(), True),
    StructField("Assists", FloatType(), True),
    StructField("Deaths", FloatType(), True),
    StructField("Headshots", FloatType(), True),
    StructField("AverageDamagePerRound", FloatType(), True),
    StructField("Kast", FloatType(), True),
    StructField("Rating", FloatType(), True),
    StructField("EntryKills", FloatType(), True),
    StructField("QuadKills", FloatType(), True),
    StructField("Aces", FloatType(), True),
    StructField("Clutch1v2s", FloatType(), True),
    StructField("Clutch1v3s", FloatType(), True),
    StructField("Clutch1v4s", FloatType(), True),
    StructField("Clutch1v5s", FloatType(), True),
])

map_schema = StructType([
    StructField("Number", IntegerType(), True),
    StructField("Name", StringType(), True),
    StructField("Status", StringType(), True),
    StructField("CurrentRound", IntegerType(), True),
    StructField("TeamAScore", IntegerType(), True),
    StructField("TeamBScore", IntegerType(), True),
])

SCHEMAS = {
  "tb_leaderboards": leaderboards_schema,
  "tb_maps": map_schema
}

if "spark-warehouse" not in os.listdir():
  spark.sql("CREATE DATABASE bronze")

try:
  for TABLE_NAME in list(SCHEMAS.keys()):
    '''
    Full load
    '''
    if TABLE_NAME not in os.listdir('spark-warehouse/bronze.db'):
      df = spark.read.parquet(f"raw/{TABLE_NAME}")
      windowSpec = window.Window.partitionBy("GameId").orderBy("UpdatedUtc") # .orderBy(1)
      df_new = df.withColumn("row_number", F.row_number().over(windowSpec)).filter("row_number = 1").drop("row_number")
      df_new.write.mode("overwrite").format("delta").saveAsTable(f"bronze.{TABLE_NAME}") # overwriting it's not overwrititng because it creates a different file name
      # df_new.write.format("delta").saveAsTable(name=f"{warehouse_location}.bronze.{TABLE_NAME}", mode="overwrite")
      # df_new.write.mode("overwrite").format("delta").saveAsTable(f"bronze.{TABLE_NAME}")

    bronzeDeltaTable = delta.tables.DeltaTable.forPath(spark, f"spark-warehouse/bronze.db/{TABLE_NAME}") #"bronze"

    '''
    When new matches lands in raw, a stream is responsible for saving these new matches in bronze.
    '''
    df_stream = ( spark.readStream
                      .format("parquet")
                      .schema(SCHEMAS[TABLE_NAME])
                      .load(f"raw/{TABLE_NAME}")
                )

    stream = ( df_stream.writeStream
                        .foreachBatch(upsertToDelta)
                        .option("checkpointLocation", f"spark-warehouse/bronze.db/{TABLE_NAME}_checkpoint")
                        .outputMode("update")
                        .start()
            )

    stream.processAllAvailable()
    stream.stop()
finally:
  spark.stop()

已尝试的无效方案 #

• 在bronze前添加spark-warehouse.前缀
• 给spark-warehouse、bronze、{TABLE_NAME}添加反引号
• 使用save()替代saveAsTable()，仅能将数据保存到bronze.db文件夹，但spark.sql("USE bronze")仍报错，spark.sql("SHOW DATABASES").show()仅显示default库

环境信息 #

• 操作系统：Windows10 / Ubuntu22.04
• 依赖版本：PySpark3.3.1、Hadoop3、delta-spark2.2.0、Java11.0.16

有效解决方案 #

1. 修复数据库创建逻辑 #

原代码通过判断spark-warehouse目录是否存在来创建数据库，逻辑不严谨——若spark-warehouse已存在但bronze库未创建，会跳过创建步骤。改用Spark Catalog直接检查数据库是否存在：

# 替换原有的if判断
if not spark.catalog.databaseExists("bronze"):
    # 显式指定数据库存储位置，避免路径歧义
    spark.sql("CREATE DATABASE bronze LOCATION 'spark-warehouse/bronze.db'")

2. 确保数据库被Spark识别 #

创建数据库后，切换到该数据库再执行表操作，避免全路径的歧义：

# 在保存表之前执行
spark.sql("USE bronze")
# 此时直接使用表名即可，无需前缀
df_new.write.mode("overwrite").format("delta").saveAsTable(TABLE_NAME)

3. 验证Delta Catalog配置 #

确认SparkSession的Delta相关配置正确，原代码中的配置是正确的，但需确保初始化顺序无误：

builder = SparkSession.builder \
    .master('local[*]') \
    .config("spark.sql.warehouse.dir", warehouse_location) \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")

spark = delta.configure_spark_with_delta_pip(builder).getOrCreate()

4. 检查文件系统权限（仅Linux/Ubuntu） #

确保Spark进程对spark-warehouse目录有读写权限，可执行以下命令调整：

sudo chown -R $USER:$USER spark-warehouse
sudo chmod -R 755 spark-warehouse

5. 显式注册已存在的表（如果用save()保存过） #

若已用save()将数据保存到bronze.db目录，可通过以下命令将表注册到Spark元数据：

spark.sql("CREATE TABLE bronze.tb_leaderboards USING DELTA LOCATION 'spark-warehouse/bronze.db/tb_leaderboards'")
spark.sql("CREATE TABLE bronze.tb_maps USING DELTA LOCATION 'spark-warehouse/bronze.db/tb_maps'")

问题根源 #

• 原代码依赖文件系统目录判断数据库是否需要创建，逻辑不严谨，导致数据库未实际创建
• 使用save()仅将数据写入文件系统，不会在Spark元数据中注册数据库和表，因此无法通过Spark SQL识别

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