Livy部署及使用文档

2020-04-23

前言

本次部署是在单台机器10.57.17.215上部署，只部署了spark和livy；通过配置livy加载本地文件jar；然后通过livy rest api提交spark任务；

一、前置条件

1.安装spark: (部署参考：https://zyongjava.github.io/2019/12/24/Spark%E9%9B%86%E7%BE%A4%E9%83%A8%E7%BD%B2/)
2.安装hadoop(可选，没有使用本地文件测试)

二、安装部署

1）下载安装包

http://livy.incubator.apache.org/download/

2）创建配置文件livy.conf

1
2
3

cd apache-livy-0.7.0-incubating-bin/conf
cp livy.conf.template livy.conf
vi livy.conf

livy.conf中修改内容如下：

# What host address to start the server on. By default, Livy will bind to all network interfaces.
livy.server.host = 10.57.17.215

# What spark master Livy sessions should use.
livy.spark.master = spark://10.57.17.215:7077

# List of local directories from where files are allowed to be added to user sessions. By
# default it's empty, meaning users can only reference remote URIs when starting their
# sessions.
livy.file.local-dir-whitelist = /home/admin/apps/jars/

3）创建配置文件livy-env.sh

1
2
3

cd apache-livy-0.7.0-incubating-bin/conf
cp livy-env.sh.template livy-env.sh
vi livy-env.sh

livy-env.sh中新增内容如下：

1 2	export SPARK_HOME=/usr/lib/spark export HADOOP_CONF_DIR=/etc/hadoop/conf

4）创建配置文件spark-blacklist.conf

1
2
3

cd apache-livy-0.7.0-incubating-bin/conf
cp spark-blacklist.conf.template spark-blacklist.conf
vi spark-blacklist.conf

spark-blacklist.conf中修改内容如下（注释掉下面2行内容），让客户端可以配置这2个参数：

1
2
3

# Disallow overriding the master and the deploy mode.
#spark.master
#spark.submit.deployMode

5）启动程序

1
2
3

cd apache-livy-0.7.0-incubating-bin

./bin/livy-server start

6）验证测试

1	http://10.57.17.215:8998/ui

三、REST调用

API文档地址：http://livy.incubator.apache.org/docs/latest/rest-api.html

1）调用测试类

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * https://www.jianshu.com/p/65d3150e8999
 *
 * https://blog.csdn.net/qq_36330643/article/details/76686225?locationNum=1&fps=1
 *
 * Q: requirement failed: Local path /data01/admin/apps/zy/transform/lib/etl.jar cannot be added to user sessions.
 *
 * A: 修改livy.conf： livy.file.local-dir-whitelist = /data01/admin/jars/
 *
 * # List of local directories from where files are allowed to be added to user sessions. By
 # default it's empty, meaning users can only reference remote URIs when starting their
 # sessions.
 *
 *
 * @author: zhengyong Date: 2020/4/17 Time: 9:48 AM
 */
public class LivyTest {

    public static void main(String[] args) throws IOException {
        String url = "http://10.57.17.215:8998/batches";
        List<Integer> sessionIds = getBatches(url);
        deleteBatches(url, sessionIds);
        String batches = postBatches(url);
    }


    public static void deleteBatches(String url, List<Integer> sessionIds) throws IOException {
        for (Integer id : sessionIds) {
            String deleteResult = HttpConfigClient.deleteLivy(String.format("%s/%s", url, id));
            System.out.println(String.format("delete batches id=%s, result: %s", id, deleteResult));
        }
    }


    public static List<Integer> getBatches(String url) throws IOException {
        String sessions = HttpConfigClient.getLivy(url, null);
        JSONObject object = JSON.parseObject(sessions);
        JSONArray array = object.getJSONArray("sessions");
        if (array == null) {
            return null;
        }
        List<Integer> sessionIds = new ArrayList<>(array.size());
        for (Object o : array) {
            JSONObject obj = (JSONObject) o;
            int id = obj.getIntValue("id");
            String name = obj.getString("name");
            String state = obj.getString("state");
            sessionIds.add(id);
            System.out.println(String.format("query session id=%s, name=%s(%s)", id, name, state));
        }
        return sessionIds;
    }

    public static String postBatches(String url) throws IOException {
        LivyEntity livyEntity = new LivyEntity();
        livyEntity.setFile("/home/admin/apps/jars/etl.jar");
        livyEntity.setClassName("cn.etl.DataEtlMain");
        livyEntity.setNumExecutors(3);
        livyEntity.setName("ETL");
        livyEntity.setExecutorMemory("4g");
        livyEntity.setExecutorCores(2);
        livyEntity.setArgs(Lists.newArrayList("params1"));

        //spark参数设置： http://spark.apache.org/docs/latest/configuration.html
        Map<String, Object> conf = new HashMap<>();
        conf.put("spark.master","spark://10.57.17.215:7077");
        conf.put("spark.submit.deployMode", "cluster");   // 这一行可以把任务在spark UI上面显示；但是jar包log不能显示
        conf.put("spark.driver.cores", "2");
        conf.put("spark.driver.memory", "4g");
        livyEntity.setConf(conf);

        String result = HttpConfigClient.postLivy(url, JSON.toJSONString(livyEntity));
        System.out.println("post batches result: " + result);
        return result;
    }
}

2）httpclient工具类

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpDelete;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;

/**
 *
 */
public class HttpConfigClient {
    
    public static String postLivy(String postUrl, String json) throws IOException {
        CloseableHttpClient livyClient = HttpClients.createDefault();
        HttpPost httpPost = new HttpPost(postUrl);
        StringEntity entity = new StringEntity(json, ContentType.APPLICATION_JSON);
        httpPost.setEntity(entity);
        CloseableHttpResponse response = livyClient.execute(httpPost);
        return handleResponse(response);
    }

    public static String getLivy(String getUrl, Map<String, Object> parametersMap) throws IOException {
        CloseableHttpClient livyClient = HttpClients.createDefault();
        String url = buildUri(getUrl, parametersMap).toString();
        HttpGet httpGet = new HttpGet(url);
        CloseableHttpResponse response = livyClient.execute(httpGet);
        return handleResponse(response);
    }

    public static String deleteLivy(String deleteUrl) throws IOException {
        CloseableHttpClient livyClient = HttpClients.createDefault();
        HttpDelete httpDelete = new HttpDelete(deleteUrl);
        CloseableHttpResponse response = livyClient.execute(httpDelete);
        return handleResponse(response);
    }

    /***
     * 拼接get请求调用的URL
     *
     * @param url 调用请求
     * @param parametersMap 业务数据，作为参数
     * @return 完成url
     */
    private static URI buildUri(String url, Map<String, Object> parametersMap) {
        if (MapUtils.isEmpty(parametersMap)) {
            return URI.create(url);
        }
        ArrayList list = new ArrayList(parametersMap.size());
        Iterator iterator = parametersMap.entrySet().iterator();

        while (iterator.hasNext()) {
            Map.Entry entry = (Map.Entry) iterator.next();
            list.add(entry.getKey().toString().trim() + "=" + entry.getValue().toString().trim());
        }
        return list.isEmpty() ? URI.create(url) : URI.create(url + "?" + StringUtils.join(list, "&"));
    }

    /**
     * 获取结果
     *
     * @param response 响应
     * @return 返回结果
     * @throws IOException
     */
    private static String handleResponse(CloseableHttpResponse response) throws IOException {
        HttpEntity resultEntity = response.getEntity();
        return EntityUtils.toString(resultEntity);
    }
}

3）实体对象

import java.util.List;
import java.util.Map;

/**
 * Livy REST API 封装
 * 批处理任务 属性封装
 */
public class LivyEntity {


    /**

     * 必须有
     * 包含需要执行应用的文件，主要是jar包
     */
    private String file;
    /**
     * User to impersonate when running the job
     */
    private String proxyUser;
    /**
     * Application Java/Spark main class
     * 主类
     */
    private String className;

    /**
     * Command line arguments for the application
     * 参数
     */
    private List<String> args;

    /**
     * jars to be used in this session
     * 这个任务里面用到的其他 jar 包
     */
    private List<String> jars;

    /**
     * Python files to be used in this session
     */
    private List<String> pyFiles;
    /**
     * files to be used in this session
     */
    private List<String> files;

    /**
     * Amount of memory to use for the driver process
     */
    private String driverMemory;

    /**
     * Number of cores to use for the driver process
     */
    private Integer driverCores;

    /**
     * Amount of memory to use per executor process
     */
    private String executorMemory;
    /**
     * Number of cores to use for each executor
     */
    private Integer executorCores;
    /**
     * Number of executors to launch for this session
     */
    private Integer numExecutors;
    /**
     * Archives to be used in this session
     */
    private List<String> archives;
    /**
     * The name of the YARN queue to which submitted
     */
    private String queue;
    /**
     * The name of this session
     * 任务名称
     */
    private String name;
    /**
     * Spark configuration properties
     * spark 配置文件
     */
    private Map<String, Object> conf;

    public String getFile() {
        return file;
    }

    public void setFile(String file) {
        this.file = file;
    }

    public String getProxyUser() {
        return proxyUser;
    }

    public void setProxyUser(String proxyUser) {
        this.proxyUser = proxyUser;
    }

    public String getClassName() {
        return className;
    }

    public void setClassName(String className) {
        this.className = className;
    }

    public List<String> getArgs() {
        return args;
    }

    public void setArgs(List<String> args) {
        this.args = args;
    }

    public List<String> getJars() {
        return jars;
    }

    public void setJars(List<String> jars) {
        this.jars = jars;
    }

    public List<String> getPyFiles() {
        return pyFiles;
    }

    public void setPyFiles(List<String> pyFiles) {
        this.pyFiles = pyFiles;
    }

    public List<String> getFiles() {
        return files;
    }

    public void setFiles(List<String> files) {
        this.files = files;
    }

    public String getDriverMemory() {
        return driverMemory;
    }

    public void setDriverMemory(String driverMemory) {
        this.driverMemory = driverMemory;
    }

    public Integer getDriverCores() {
        return driverCores;
    }

    public void setDriverCores(Integer driverCores) {
        this.driverCores = driverCores;
    }

    public String getExecutorMemory() {
        return executorMemory;
    }

    public void setExecutorMemory(String executorMemory) {
        this.executorMemory = executorMemory;
    }

    public Integer getExecutorCores() {
        return executorCores;
    }

    public void setExecutorCores(Integer executorCores) {
        this.executorCores = executorCores;
    }

    public Integer getNumExecutors() {
        return numExecutors;
    }

    public void setNumExecutors(Integer numExecutors) {
        this.numExecutors = numExecutors;
    }

    public List<String> getArchives() {
        return archives;
    }

    public void setArchives(List<String> archives) {
        this.archives = archives;
    }

    public String getQueue() {
        return queue;
    }

    public void setQueue(String queue) {
        this.queue = queue;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Map<String, Object> getConf() {
        return conf;
    }

    public void setConf(Map<String, Object> conf) {
        this.conf = conf;
    }

}

四、异常问题

问题：requirement failed: Local path /home/admin/apps/jars/etl.jar cannot be added to user sessions.

解决办法：修改livy.conf配置

1	livy.file.local-dir-whitelist = /home/admin/apps/jars/

五、代码地址

https://github.com/zyongjava/pomelo/tree/master/src/main/java/livy