Spark 2.4.1读取Kafka数据时遇ClassNotFoundException求助

错误根源 #

java.lang.ClassNotFoundException: org.apache.spark.sql.kafka010.KafkaSourceRDDPartition 报错的核心原因是：Executor节点（spark-worker容器）缺少Kafka数据源的依赖类。当你将从Kafka读取的DataFrame转换为RDD时，这个RDD的分区类型是KafkaSourceRDDPartition，属于Spark Kafka扩展包，但你的Worker容器并未加载该依赖——尽管你用spark-submit --packages指定了依赖，但在分布式集群环境下，该参数仅会将包传递给Driver节点，Worker节点无法自动获取到依赖（尤其是自定义Docker镜像未预装依赖的情况下）。

解决方案 #

方案1：将Kafka依赖预装到所有Spark节点的Docker镜像中 #

修改自定义Spark镜像的构建脚本，提前把Spark Kafka扩展包及其依赖安装到Spark的jars目录，或者通过配置文件设置全局依赖：

1. 在Dockerfile中添加以下命令（基于Apache Spark基础镜像）：

# 下载并安装Spark Kafka相关依赖包
RUN cd $SPARK_HOME/jars && \
    curl -O https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.11/2.4.1/spark-sql-kafka-0-10_2.11-2.4.1.jar && \
    curl -O https://repo1.maven.org/maven2/org/apache/spark/spark-streaming-kafka-0-10_2.11/2.4.1/spark-streaming-kafka-0-10_2.11-2.4.1.jar && \
    curl -O https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/2.0.0/kafka-clients-2.0.0.jar

2. 或者在spark-defaults.conf中添加全局依赖配置：

spark.jars.packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.1

重新构建镜像并重启所有Spark容器，确保Master和Worker节点都能加载到Kafka依赖。

方案2：修改代码避免直接操作Kafka源RDD #

调整代码逻辑，绕过对KafkaSourceRDD的依赖，将Kafka读取到的value列转换为普通字符串RDD：

def infer_topic_schema_json(schema_topic, schema_inference_sample_size):
    df_json = (spark.read.format("kafka")
            .option("kafka.bootstrap.servers", "broker:29092")
            .option("subscribe", schema_topic)
            .option("startingOffsets", "earliest")
            .option("maxOffsetsPerTrigger", schema_inference_sample_size)
            .option("failOnDataLoss", "false")
            .load()
            .withColumn("value", F.expr("string(value)"))
            .select("value"))
    
    # 先将数据收集到Driver端，再转换为普通字符串RDD（注意：sample_size过大时会占用Driver内存）
    json_samples = df_json.select("value").rdd.map(lambda x: x.value).collect()
    # 用普通字符串RDD推断Schema，避免Kafka相关类依赖
    df_read = spark.read.json(spark.sparkContext.parallelize(json_samples), multiLine=True)
    
    return df_read.schema.json()

如果schema_inference_sample_size较大，可改用临时文件中转的方式：

def infer_topic_schema_json(schema_topic, schema_inference_sample_size):
    df_json = (spark.read.format("kafka")
            .option("kafka.bootstrap.servers", "broker:29092")
            .option("subscribe", schema_topic)
            .option("startingOffsets", "earliest")
            .option("maxOffsetsPerTrigger", schema_inference_sample_size)
            .option("failOnDataLoss", "false")
            .load()
            .withColumn("value", F.expr("string(value)"))
            .select("value"))
    
    # 将JSON字符串写入临时目录
    temp_path = "/tmp/spark_schema_temp"
    df_json.write.mode("overwrite").text(temp_path)
    # 从临时目录读取并推断Schema
    df_read = spark.read.json(temp_path, multiLine=True)
    
    return df_read.schema.json()

方案3：确保Spark Submit时依赖包分发到所有节点 #

在spark-submit命令中添加额外配置，强制将依赖包分发到Executor节点：

docker exec -it my-container's-id spark-submit \
    --packages 'org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.1' \
    --conf spark.driver.extraClassPath=$(echo $SPARK_HOME/jars/*.jar | tr ' ' ':') \
    --conf spark.executor.extraClassPath=$(echo $SPARK_HOME/jars/*.jar | tr ' ' ':') \
    /app/spark_processor.py

或者通过spark.jars.packages配置强制所有节点加载依赖：

docker exec -it my-container's-id spark-submit \
    --conf spark.jars.packages=org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.1 \
    /app/spark_processor.py

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