PySpark整合Kafka报错：找不到Kafka数据源问题求助

核心原因 #

这个错误本质是Spark未加载到官方Kafka连接器依赖包，即便配置了环境变量，Colab的PySpark初始化逻辑可能忽略手动添加的依赖，或是依赖版本不匹配导致加载失败。

针对Colab的有效解决方案 #

1. 初始化SparkSession时指定正确的Kafka连接器依赖 #

在Colab中，不要单独安装依赖后启动PySpark，直接在初始化SparkSession时通过spark.jars.packages参数指定匹配版本的连接器。Spark 3.5.0对应的连接器版本为org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0（注意：连接器的Scala版本需与Spark一致，和你安装的Kafka的Scala版本无关）。

示例代码：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("KafkaStreamProcessing") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
    .getOrCreate()

2. 清理Colab缓存的旧依赖 #

若之前安装过不匹配的依赖，Colab可能缓存旧包，执行以下命令清理：

!rm -rf ~/.ivy2/cache/org.apache.spark/

清理后重新运行SparkSession初始化代码，让系统重新下载正确依赖。

3. 验证Kafka数据源是否加载成功 #

初始化完成后，运行以下代码检查：

# 查看已注册的数据源
spark.sql("SHOW DATA SOURCES").show()

若输出包含kafka，说明依赖加载成功。

4. 确认Kafka服务的连通性 #

若依赖加载成功仍报错，检查Colab能否访问你的Kafka集群：

# 替换为你的Kafka broker地址和端口，测试连通性
!telnet your-kafka-broker-ip 9092

若连通失败，需确保Kafka集群允许Colab的IP访问，或使用公网可访问的Kafka服务。

常见误区 #

• 不要混淆Kafka的Scala版本与Spark连接器的Scala版本：连接器的Scala版本必须与Spark本身一致（Spark 3.5.0默认使用Scala 2.12），和你安装的Kafka的Scala版本（2.13）无关。
• 不要手动下载jar包后添加到环境变量：Colab的PySpark初始化优先使用自身依赖管理，手动添加的jar包可能无法被加载。

内容的提问来源于stack exchange，提问作者Nícolas Farfán Cheneaux