前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >如何通过Livy的RESTful API接口向非Kerberos环境的CDH集群提交作业

如何通过Livy的RESTful API接口向非Kerberos环境的CDH集群提交作业

作者头像
Fayson
修改2018-04-01 21:06:55
2.4K0
修改2018-04-01 21:06:55
举报
文章被收录于专栏:Hadoop实操Hadoop实操

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。 Fayson的github:https://github.com/fayson/cdhproject 提示:代码块部分可以左右滑动查看噢

1.文档编写目的


在前面的文章Fayson介绍了《Livy,基于Apache Spark的开源REST服务,加入Cloudera Labs》和《如何编译Livy并在非Kerberos环境的CDH集群中安装》,Livy提供了两种类型的API(编程API和RESTful API接口),本篇文章主要介绍如何使用java代码调用Livy提供的RESTful API接口向非Kerberos环境的CDH集群提交Spark作业操作。

  • 内容概述

1.开发环境准备

2.Livy调用示例代码

3.示例代码运行及验证

  • 测试环境

1.CM和CDH版本为5.13.1

2.Livy版本为0.4

  • 前置条件

1.集群未启用Kerberos

2.环境准备及描述


1.我们将作业运行的jar包上传到HDFS目录

这里Fayson使用的Spark自带的示例来测试。

2.使用Maven创建Livy示例工程

3.在pom文件中添加如下依赖

代码语言:txt
复制
<dependency>
    <groupId>org.apache.httpcomponents</groupId>
    <artifactId>httpclient</artifactId>
    <version>4.5.4</version>
</dependency>

3.编写示例代码


1.HTTP请求的工具类(HttpUtils.java)

代码语言:txt
复制
package com.cloudera.utils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
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.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.util.Map;
/**
 * package: com.cloudera
 * describe: 封装非Kerberos环境的Http请求工具类
 * creat_user: Fayson
 * email: htechinfo@163.com
 * creat_date: 2018/2/12
 * creat_time: 下午12:16
 * 公众号:Hadoop实操
 */
public class HttpUtils {
    /**
     * HttpGET请求
     * @param url
     * @param headers
     * @return
     */
    public static String getAccess(String url, Map<String,String> headers) {
        String result = null;
        CloseableHttpClient httpClient = HttpClients.createDefault();
        HttpGet httpGet = new HttpGet(url);
        if(headers != null && headers.size() > 0){
            headers.forEach((K,V)->httpGet.addHeader(K,V));
        }
        try {
            HttpResponse response = httpClient.execute(httpGet);
            HttpEntity entity = response.getEntity();
            result = EntityUtils.toString(entity);
            System.out.println(result);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }
    /**
     * HttpDelete请求
     * @param url
     * @param headers
     * @return
     */
    public  static String deleteAccess(String url, Map<String,String> headers) {
        String result = null;
        CloseableHttpClient httpClient = HttpClients.createDefault();
        HttpDelete httpDelete = new HttpDelete(url);
        if(headers != null && headers.size() > 0){
            headers.forEach((K,V)->httpDelete.addHeader(K,V));
        }
        try {
            HttpResponse response = httpClient.execute(httpDelete);
            HttpEntity entity = response.getEntity();
            result = EntityUtils.toString(entity);
            System.out.println(result);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return result;
    }
    /**
     * HttpPost请求
     * @param url
     * @param headers
     * @param data
     * @return
     */
    public static String postAccess(String url, Map<String,String> headers, String data)  {
        String result = null;
        CloseableHttpClient httpClient = HttpClients.createDefault();
        HttpPost post = new HttpPost(url);
        if(headers != null && headers.size() > 0){
            headers.forEach((K,V)->post.addHeader(K,V));
        }
        try {
            StringEntity entity = new StringEntity(data);
            entity.setContentEncoding("UTF-8");
            entity.setContentType("application/json");
            post.setEntity(entity);
            HttpResponse response = httpClient.execute(post);
            HttpEntity resultEntity = response.getEntity();
            result = EntityUtils.toString(resultEntity);
            System.out.println(result);
            return result;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }
}

2.Livy RESTful API调用示例代码

代码语言:txt
复制
package com.cloudera.nokerberos;
import com.cloudera.utils.HttpUtils;
import java.util.HashMap;
/**
 * package: com.cloudera
 * describe: 通过Java代码调用Livy的RESTful API实现向非Kerberos的CDH集群作业提交
 * creat_user: Fayson
 * email: htechinfo@163.com
 * creat_date: 2018/2/11
 * creat_time: 上午10:50
 * 公众号:Hadoop实操
 */
public class AppLivy {
    private static String LIVY_HOST = "http://ip-172-31-7-172.fayson.com:8998";
    public static void main(String[] args) {
        HashMap<String, String> headers = new HashMap<>();
        headers.put("Content-Type", "application/json");
        headers.put("Accept", "application/json");
        headers.put("X-Requested-By", "fayson");
        //创建一个交互式会话
//        String kindJson = "{\"kind\": \"spark\", \"proxyUser\":\"fayson\"}";
//        HttpUtils.postAccess(LIVY_HOST + "/sessions", headers, kindJson);
        //执行code
//        String code = "{\"code\":\"sc.parallelize(1 to 2).count()\"}";
//        HttpUtils.postAccess(LIVY_HOST + "/sessions/1/statements", headers, code);
        //删除会话
//        HttpUtils.deleteAccess(LIVY_HOST + "/sessions/2", headers);
        //封装提交Spark作业的JSON数据
        String submitJob = "{\"className\": \"org.apache.spark.examples.SparkPi\",\"executorMemory\": \"1g\",\"args\": [200],\"file\": \"/fayson-yarn/jars/spark-examples-1.6.0-cdh5.13.1-hadoop2.6.0-cdh5.13.1.jar\", \"proxyUser\":\"fayson\"}";
        //向集群提交Spark作业
        HttpUtils.postAccess(LIVY_HOST + "/batches", headers, submitJob);
        //通过提交作业返回的SessionID获取具体作业的执行状态及APPID
        HttpUtils.getAccess(LIVY_HOST + "/batches/3", headers);
    }
}

4.示例代码运行


1.运行AppLivy代码,向集群提交Spark作业

响应结果:

代码语言:txt
复制
{
  "id": 4,
  "state": "starting",
  "appId": null,
  "appInfo": {
    "driverLogUrl": null,
    "sparkUiUrl": null
  },
  "log": ["stdout: ", "\nstderr: ", "\nYARN Diagnostics: "]
}

2.获取作业运行状态,将上一步获取到的id传入到如下请求

响应结果:

代码语言:txt
复制
{
  "id": 4,
  "state": "success",
  "appId": "application_1518401384543_0008",
  "appInfo": {
    "driverLogUrl": null,
    "sparkUiUrl": "http://ip-172-31-6-148.fayson.com:8088/proxy/application_1518401384543_0008/"
  },
  "log": ["stdout: ", "WARNING: User-defined SPARK_HOME (/opt/cloudera/parcels/CDH-5.13.1-1.cdh5.13.1.p0.2/lib/spark) overrides detected (/opt/cloudera/parcels/CDH/lib/spark).", "WARNING: Running spark-class from user-defined location.", "\nstderr: ", "\nYARN Diagnostics: "]
}

通过如上返回的结果,我们可以看到作业的APPID。

3.查看Livy界面提交作业的状态

4.通过CM和Yarn的8088界面查看作业执行结果

Yarn的应用程序界面显示

显示作业运行成功

GitHub地址:

https://github.com/fayson/cdhproject/blob/master/livydemo/src/main/java/com/cloudera/nokerberos/AppLivy.java

https://github.com/fayson/cdhproject/blob/master/livydemo/src/main/java/com/cloudera/utils/HttpUtils.java

提示:代码块部分可以左右滑动查看噢 为天地立心,为生民立命,为往圣继绝学,为万世开太平。 温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。

推荐关注Hadoop实操,第一时间,分享更多Hadoop干货,欢迎转发和分享。

原创文章,欢迎转载,转载请注明:转载自微信公众号Hadoop实操

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2018-02-14,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Hadoop实操 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
专用宿主机
专用宿主机(CVM Dedicated Host,CDH)提供用户独享的物理服务器资源,满足您资源独享、资源物理隔离、安全、合规需求。专用宿主机搭载了腾讯云虚拟化系统,购买之后,您可在其上灵活创建、管理多个自定义规格的云服务器实例,自主规划物理资源的使用。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档