PySpark报错NoClassDefFoundError: kafka/common/TopicAndPartition

该问题属于依赖配置错误，和PySpark流处理的代码逻辑无关。

• 核心报错java.lang.NoClassDefFoundError: kafka/common/TopicAndPartition的直接原因是类路径缺失Kafka 0.8版本原生客户端类：你配置的spark-streaming-kafka-0-8_2.11-2.3.0.jar只是Spark对Kafka 0.8的适配层包，本身不包含Kafka原生客户端的类；同时你额外引入的spark-sql-kafka-0-10_2.11-2.3.0.jar是对接Kafka 0.10+版本的连接器，两个不同大版本的Kafka连接器共存会触发类路径冲突，打乱类加载优先级，进一步导致0.8版本的Kafka核心类无法被正常加载。
• 后续抛出的Py4JNetworkError、createDirectStreamWithMessageHandler调用失败属于连带故障：JVM端的ApplicationMaster因为类缺失提前退出，Python端和JVM建立的Py4J连接直接断开，才会触发网络类报错，不是createDirectStream对接逻辑本身的代码写法问题。

• 清理冲突依赖：如果你的代码是用DStream的createDirectStream对接Kafka 0.8集群，直接从spark.jars.packages配置中移除spark-sql-kafka-0-10_2.11-2.3.0.jar，禁止0.8和0.10两个版本的Kafka连接器同时加载。
• 补全缺失依赖：在spark.jars.packages中补充和你集群Kafka版本完全一致的0.8.x系列Kafka客户端包，注意包的Scala版本必须和Spark 2.3.0自带的Scala 2.11版本对齐，例如对接Kafka 0.8.2.2集群就添加kafka_2.11:0.8.2.2。你之前引入的metrics-core-2.2.0.jar可以保留，不存在冲突。
• 调整类加载优先级：通过Livy提交作业时，额外添加两项配置：spark.driver.userClassPathFirst=true、spark.executor.userClassPathFirst=true，避免Ambari集群预置的高版本Kafka客户端包优先加载，覆盖你提交的0.8版本依赖类。
• 提交校验：配置调整后重新提交作业，只要stderr日志中不再出现kafka/common/TopicAndPartition类找不到的报错，Py4J连接异常的连带问题会自动消失，流作业即可正常运行。

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