Spark Structured Streaming提交含JmsWriter的Jar仍报ClassNotFoundException如何解决

问题根因 #

你遇到的ClassNotFoundException是Executor侧反序列化JmsWriter对象时，无法在自身类路径下找到对应类导致的，同时你的JmsWriter实现还存在隐藏的序列化问题，具体原因和解决方法如下：

解决方法 #

1. 确保自定义Jar分发到所有Executor #

Spark运行时Driver和Executor使用独立的类加载器，仅将Jar作为spark-submit的主Jar参数仅会将其加载到Driver的类路径，Executor无法访问：

• 提交任务时添加--jars参数指定你的Jar路径，Spark会自动将该Jar分发到所有Executor的类路径下：

  spark-submit \
  --class 你的主类全限定名 \
  --jars output-jms.jar \
  # 其他你的任务参数
  output-jms.jar
  

• 如果运行在YARN/K8s集群环境，建议将Jar上传到所有节点都可访问的共享存储（如HDFS），--jars参数填写HDFS路径即可。
• 也可以将项目打成包含所有业务依赖（ActiveMQ相关依赖等）的Fat Jar，注意Spark、Scala相关依赖要设置为provided避免和集群内置依赖冲突，直接提交Fat Jar即可无需额外加--jars。

2. 修复JmsWriter的序列化问题 #

即使解决了类找不到的问题，你当前的JmsWriter代码也会报序列化错误：
jmsConnectionFactory字段未加@transient注解，而ActiveMQ的连接工厂是不可序列化的，Spark序列化JmsWriter对象分发到Executor时会失败，需要修改代码：

class JmsWriter(username: String, password: String, brokerUrl: String, clientId: String, topic: String) extends ForeachWriter[Row] with Logging {

  // 加@transient注解避免序列化，改为lazy在open方法调用时才初始化
  @transient lazy val jmsConnectionFactory = JmsConnector.initActiveMqConnectionFactory(username, password, brokerUrl, clientId)
  @transient var connection: Connection = _
  @transient var jmsSession: Session = _
  @transient var jmsProducer: MessageProducer = _

  override def open(partitionId: Long, epochId: Long): Boolean = {
    connection = {
      val cn = jmsConnectionFactory.createConnection()
      cn.setClientID(clientId)
      cn.start()
      logInfo("ActiveMQ connection created")
      cn
    }
    jmsSession = JmsConnector.initActiveMqSession(connection)
    jmsProducer = JmsConnector.initActiveMqProducer(jmsSession, topic)
    true
  }

  override def process(value: Row): Unit = jmsProducer.send(jmsSession.createTextMessage(value.mkString("")))

  override def close(errorOrNull: Throwable): Unit = {
    if (errorOrNull != null) logError("Error during writing the stream: " + errorOrNull)
    jmsProducer.close()
    jmsSession.close()
    connection.close()
  }
}

3. 验证类是否正常打包 #

先解压你的output-jms.jar检查是否存在对应类文件：

jar tf output-jms.jar | grep com.addmeaning.output.jms.spark.service.JmsWriter

如果没有输出，说明打包时没有将该类打入Jar，需要修改你的构建配置（maven/gradle/sbt）确保编译后的类正常打入Jar包。

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