Spark 3.3.0连接Kafka 2.4.1失败：数据源找不到错误求助

1. 确认是否添加了Spark Kafka专属连接器依赖 #

你之前只添加了kafka-clients，但Spark连接Kafka需要专门的Spark Kafka连接器包——这个包才是提供kafka数据源的核心依赖，kafka-clients只是底层客户端库，单独添加无法解决数据源找不到的问题。

Spark 3.3.0对应的连接器版本为3.3.0，且需匹配Scala版本（Spark 3.3.0默认使用Scala 2.12），正确的依赖坐标是org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0。

两种添加方式： #

• 启动PySpark时指定：
  在终端执行启动命令时带上--packages参数：

  pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0
  

  若要启动Jupyter Notebook，使用以下命令：

  PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=notebook pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0
  

• 在SparkSession中配置：
  手动创建SparkSession时添加依赖配置：

  from pyspark.sql import SparkSession
  spark = SparkSession.builder \
      .appName("KafkaConnectorTest") \
      .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0") \
      .getOrCreate()
  

2. 验证依赖是否成功加载 #

在Jupyter中执行以下代码，查看已加载的Jar包列表，确认包含spark-sql-kafka-0-10_2.12-3.3.0.jar和对应的kafka-clients包：

for jar in spark.sparkContext.listJars():
    print(jar)

如果未看到相关Jar，说明依赖未加载成功：检查网络是否能访问Maven仓库，或者手动下载Jar包放到Spark的jars目录下。

3. 核对Kafka版本兼容性与依赖冲突 #

Spark 3.3.0的Kafka连接器支持Kafka 0.10.0+，你的Kafka 2.4.1是兼容的，但如果连接器自带的kafka-clients版本和集群版本不一致引发问题，可以强制指定Kafka 2.4.1的客户端包：

spark = SparkSession.builder \
    .appName("KafkaConnectorTest") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0,org.apache.kafka:kafka-clients:2.4.1") \
    .config("spark.jars.excludes", "org.apache.kafka:kafka-clients") \
    .getOrCreate()

4. 检查代码中的数据源格式是否正确 #

确保读取Kafka时的format参数是"kafka"，避免拼写错误或使用旧的全类名（全类名虽能工作，但更易出错）。示例正确代码：

流处理读取 #

df_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "你的Kafka地址:9092") \
    .option("subscribe", "目标topic") \
    .load()

批处理读取 #

df_batch = spark.read \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "你的Kafka地址:9092") \
    .option("subscribe", "目标topic") \
    .option("startingOffsets", "earliest") \
    .load()

5. Jupyter环境的特殊排查 #

如果是在已启动的Jupyter内核中修改依赖配置，必须重启内核才能生效——SparkSession一旦创建，依赖无法动态添加，需重新启动Notebook内核确保依赖在Spark初始化时被加载。

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