本地运行PySpark(Kafka转Delta)报错：找不到spark_catalog插件类

问题重现 #

示例代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from delta import *

spark = SparkSession \
        .builder \
        .appName("test") \
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.jars.packages", "io.delta:delta-core_2.12:2.1.0") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
        .getOrCreate()

kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "demo.topic") \
    .option("startingOffsets", "latest") \
    .load() \
    .withColumn("current_timestamp", unix_timestamp()) \
    .withColumn("value_str", col("value").cast(StringType())) \
    .select("current_timestamp", "value_str")

stream = kafka_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "./data/tmp/delta/events/_checkpoints/") \
    .toTable("events")

stream.awaitTermination()

执行环境与命令 #

• Spark版本：3.3.1
• 执行命令：

spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0 kafka_to_delta.py

报错信息 #

Traceback (most recent call last):
  File "/Users/user/Desktop/python-module/kafka_to_delta.py", line 24, in <module>
    stream = kafka_df.writeStream \
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/pyspark/python/lib/pyspark.zip/pyspark/sql/streaming.py", line 1468, in toTable
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/pyspark/python/lib/py4j-0.10.9.5-src.zip/py4j/java_gateway.py", line 1321, in __call__
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/pyspark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 190, in deco
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/site-packages/pyspark/python/lib/py4j-0.10.9.5-src.zip/py4j/protocol.py", line 326, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o63.toTable.
: org.apache.spark.SparkException: Cannot find catalog plugin class for catalog 'spark_catalog': org.apache.spark.sql.delta.catalog.DeltaCatalog
    at org.apache.spark.sql.errors.QueryExecutionErrors$.catalogPluginClassNotFoundForCatalogError(QueryExecutionErrors.scala:1638)
    at org.apache.spark.sql.connector.catalog.Catalogs$.load(Catalogs.scala:65)
    ...
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.delta.catalog.DeltaCatalog
    at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:445)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:587)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:520)
    ...

错误原因分析 #

1. 依赖包未完整加载：spark-submit命令仅指定Kafka依赖，未包含Delta核心包，导致JVM无法找到DeltaCatalog类。代码中spark.jars.packages配置易被命令行--packages参数覆盖，无法生效。
2. 依赖版本不兼容：Kafka包版本（3.0.0）与Spark版本（3.3.1）不匹配，Spark 3.3.1对应的spark-sql-kafka-0-10包版本需为3.3.1。
3. Catalog配置依赖缺失：启用DeltaCatalog作为默认catalog的前提是Delta包正确加载，否则直接触发类找不到的错误。

修复步骤 #

1. 修正Spark Submit命令，统一依赖版本 #

将Delta和Kafka的依赖包同时加入--packages参数，确保版本与Spark 3.3.1匹配：

spark-submit \
--packages "io.delta:delta-core_2.12:2.3.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.1" \
kafka_to_delta.py

注：Delta 2.3.0是Spark 3.3.x的推荐兼容版本。

2. 优化SparkSession配置 #

移除代码中的spark.jars.packages配置，统一通过命令行管理依赖，避免冲突：

spark = SparkSession \
        .builder \
        .appName("test") \
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
        .getOrCreate()

3. 替代方案：直接写入路径（无需Catalog） #

若暂时不需要Delta Catalog的元数据管理能力，可改用save方法指定Delta表存储路径，规避Catalog相关错误：

stream = kafka_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/absolute/path/to/checkpoints") \
    .save("/absolute/path/to/delta/events")

Catalog与Schema最佳实践 #

Catalog相关 #

• 是否需要指定：如果要使用Delta Lake的ACID事务、版本回溯等元数据能力，推荐启用DeltaCatalog作为默认catalog；仅需写入Delta文件时，可直接指定路径，无需配置Catalog。
• 生产环境最佳实践：
  • 使用外部Catalog（Delta Catalog、Hive Metastore）统一管理表元数据，避免分散的路径管理。
  • 确保Catalog配置与依赖包版本严格匹配，避免兼容性问题。

Schema相关 #

• 必须显式定义：流式处理中Kafka的value字段多为结构化数据，显式定义Schema可避免自动推断导致的类型不一致、Schema漂移等问题。示例：

# 定义Kafka value的Schema
value_schema = StructType([
    StructField("id", IntegerType()),
    StructField("content", StringType()),
    StructField("created_at", TimestampType())
])

kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "demo.topic") \
    .option("startingOffsets", "latest") \
    .load() \
    .withColumn("value", from_json(col("value").cast(StringType()), value_schema)) \
    .select("value.*", unix_timestamp().alias("current_timestamp"))

• Schema漂移处理：生产环境若存在Schema变化，可启用Delta Lake的mergeSchema参数开启Schema进化，但需严格控制变更范围。

通用最佳实践 #

• 使用绝对路径存储Checkpoint和Delta表，避免相对路径在不同环境下的解析问题。
• 将Checkpoint路径配置在分布式存储（HDFS、S3等），确保任务重启后可恢复状态。
• 监控流任务的延迟与吞吐量，及时发现数据积压或异常。

内容的提问来源于stack exchange，提问作者boring-coder