容器化部署Livy时提交Spark Java作业报类找不到的解决方法

基于开源docker-spark-livy镜像部署Spark 2.4.7 + Livy 0.7服务，镜像自带的curl调用示例可正常运行，服务状态正常。开发了基于Maven构建的Java版Spark单词计数程序，尝试通过Livy原生Java客户端提交作业时触发类找不到异常，调整本地目录白名单、更换Jar包存储路径后问题仍未解决。

作业实现类 #

package spark;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.regex.Pattern;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.*;
import org.apache.livy.Job;
import org.apache.livy.JobContext;
import scala.Tuple2;

public final class JavaWordCount implements Job<Double> {

    private static final long serialVersionUID = 4870271814150948504L;
    private static final Pattern SPACE = Pattern.compile(" ");

    @Override
    public Double call(JobContext ctx) throws Exception {
        count(ctx);
        return 0.7;
    }

    public void count(JobContext ctx){
        JavaRDD<String> lines = ctx.sc().parallelize(Arrays.asList("It is close to midnight and something evil is lurking in the dark".split(" ")));

        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) {
                return Arrays.asList(SPACE.split(s)).iterator();
            }
        });

        JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) {
                return new Tuple2<>(s, 1);
            }
        });

        JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) {
                return i1 + i2;
            }
        });

        List<Tuple2<String, Integer>> output = counts.collect();
        for (Tuple2<?, ?> tuple : output) {
            System.out.println(tuple._1() + ": " + tuple._2());
        }
    }
}

客户端提交代码 #

package spark;

import org.apache.livy.Job;
import org.apache.livy.LivyClient;
import org.apache.livy.LivyClientBuilder;

import java.io.File;
import java.net.URI;

public final class SubmitJob {

    private static final String livyURI = "http://localhost:8998/";
    private static final String JAR_PATH = "/Users/.../spark-word-count/target/word-count-0.1-SNAPSHOT.jar";

    public static void main(String[] args) throws Exception {
        LivyClient livyClient = new LivyClientBuilder()
                .setURI(new URI(livyURI)).build();

        try {
            System.err.printf("Uploading %s to the Spark context...\n", JAR_PATH);
            livyClient.uploadJar(new File(JAR_PATH));

            System.err.println("Running JavaWordCount...");
            double result = livyClient.submit(new JavaWordCount()).get();

            System.out.println("Job return value: " + result);
        } finally {
            livyClient.stop(true);
        }
    }
}

执行客户端程序后抛出核心错误：

Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException:
org.apache.livy.shaded.kryo.kryo.KryoException: Unable to find class: spark.JavaWordCount
    at org.apache.livy.shaded.kryo.kryo.util.DefaultClassResolver.readName(DefaultClassResolver.java:138)
    at org.apache.livy.shaded.kryo.kryo.util.DefaultClassResolver.readClass(DefaultClassResolver.java:115)
    ...
Caused by: java.lang.ClassNotFoundException: spark.JavaWordCount

已完成的排查动作：

• 确认livy.file.local-dir-whitelist配置值为/，已放行全路径本地文件读取权限
• 将Jar包上传到Livy运行容器的根目录，修改客户端代码中的Jar路径为容器内绝对路径/word-count-0.1-SNAPSHOT.jar，重新提交后异常完全一致

经过排查，该类找不到异常和本地路径白名单配置没有关联，核心触发原因有两个：

1. uploadJar方法为异步执行逻辑，原代码调用后未等待Jar包上传、类加载完成就直接提交作业，Livy服务端还未加载到目标Jar包内的类
2. 直接将客户端本地创建的JavaWordCount实例通过Kryo序列化传输到服务端的提交方式，存在强类路径依赖，只要服务端未提前加载到对应类，就会触发反序列化类找不到错误

按以下步骤调整即可正常提交作业：

• 给Jar上传动作增加等待完成逻辑
  修改上传代码，接收uploadJar返回的异步任务对象，调用get()方法等待上传、类加载流程完全结束：

  // 替换原有上传调用，增加同步等待逻辑
  livyClient.uploadJar(new File(JAR_PATH)).get();
  

• 更换作业提交方式，避免直接传输本地作业实例
  不要在客户端代码中new JavaWordCount()传入submit方法，改为通过传入作业类全限定名的方式提交，让服务端从已加载的Jar包中自动加载类完成实例化：

  // 替换原有submit调用，使用ClassReference方式提交
  import org.apache.livy.client.common.ClassReference;
  double result = livyClient.submit(new ClassReference("spark.JavaWordCount")).get();
  

• 校验Jar包构建正确性
  执行jar tf word-count-0.1-SNAPSHOT.jar命令查看Jar包内容，确认存在spark/JavaWordCount.class文件，排除Maven打包时因依赖范围配置、Shade插件规则错误导致目标类未被打进Jar的问题。
• 补充校验上传结果
  提交作业前可调用Livy内置的文件查询接口，确认目标Jar已出现在已上传文件列表中，再触发作业提交。

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