如何向Apache Livy提交Spark作业？调整Test类适配client.submit方法

我来帮你梳理下具体的调整步骤，让你的Test类能通过client.submit(new Test())正常提交到Livy执行：

1. 让Test类实现Livy的Job接口 #

Livy的客户端提交作业时，要求作业类实现org.apache.livy.Job接口（这个接口继承了Callable，并且强制要求可序列化）。你需要把原来的Spark作业逻辑放到call()方法里，通过Livy提供的JobContext来获取Spark上下文，而不是自己创建。

示例代码结构：

import org.apache.livy.Job;
import org.apache.livy.JobContext;
import java.io.Serializable;

// 泛型参数是作业执行后返回的结果类型，不需要结果可以用Void
public class Test implements Job<String>, Serializable {

    @Override
    public String call(JobContext jobContext) throws Exception {
        // 通过JobContext获取SparkSession，Livy会帮你初始化好集群上下文
        SparkSession spark = jobContext.sparkSession();
        
        // 这里写你的Spark业务逻辑，比如读取数据、处理计算
        // 示例：读取HDFS文件并统计行数
        long count = spark.read().text("hdfs://path/to/input").count();
        
        // 返回结果，根据泛型类型调整
        return "作业执行完成，总行数：" + count;
    }
}

2. 确保类的可序列化性 #

因为Livy需要把你的Test类序列化后传输到Spark集群执行，所以：

• Test类必须实现Serializable接口（上面的示例已经包含）
• 类中的所有成员变量要么是基本数据类型，要么也实现Serializable接口
• 不要在成员变量中持有SparkSession、SparkContext这类不可序列化的对象，所有上下文都从JobContext获取

3. 调整提交代码的细节 #

提交作业时，要确保上传了包含Test类的jar包（如果你的Test类不在Livy默认加载的路径里）：

import org.apache.livy.LivyClient;
import org.apache.livy.LivyClientBuilder;
import java.io.File;
import java.net.URI;
import java.util.concurrent.Future;

public class LivySubmitExample {
    public static void main(String[] args) throws Exception {
        // 构建Livy客户端，替换成你的Livy服务器地址和端口
        try (LivyClient client = new LivyClientBuilder()
                .setURI(new URI("http://your-livy-server:8998"))
                .build()) {
            
            // 上传包含Test类的jar包到Livy服务器
            // 如果你的代码已经打包成jar，替换成实际路径
            client.uploadJar(new File("target/your-spark-app.jar")).get();
            
            // 提交Test作业
            Future<String> jobResult = client.submit(new Test());
            
            // 等待作业执行完成并获取结果
            String result = jobResult.get();
            System.out.println("作业结果：" + result);
        }
    }
}

4. 常见问题排查 #

• 序列化异常：检查Test类的成员变量，移除或替换不可序列化的对象，比如数据库连接池、未序列化的自定义类
• 类找不到异常：确认上传的jar包包含了Test类以及所有依赖的第三方类，或者通过client.addJar()添加额外依赖
• 连接超时：检查Livy服务器是否正常运行，地址和端口是否正确，防火墙是否开放8998端口

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