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

Fayson

修改于 2018-04-01 18:40:56

3.9K0

修改于 2018-04-01 18:40:56

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

1.文档编写目的

在前面的文章Fayson介绍了《Livy，基于Apache Spark的开源REST服务，加入Cloudera Labs》、《如何编译Livy并在非Kerberos环境的CDH集群中安装》、《如何通过Livy的RESTful API接口向非Kerberos环境的CDH集群提交作业》和《如何在Kerberos环境的CDH集群部署Livy》。本篇文章主要介绍如何使用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.准备访问集群的keytab及集群的krb5.conf文件

krb5.conf配置文件，获取方式：文件在CDH集群中将KDC服务器上的/etc/目录

在KDC所在服务器执行如下命令生成fayson.keytab文件

[root@ip-172-31-16-68 ~]# kadmin.local -q "xst -norandkey -k /root/fayson.keytab fayson@FAYSON.COM"

（可左右滑动）

3.JAAS文件login-yarn.conf内容如下：

Client {
    com.sun.security.auth.module.Krb5LoginModule required
    storeKey=true
    useKeyTab=true
    debug=true
    keyTab="/Volumes/Transcend/keytab/fayson.keytab"
    principal="fayson@FAYSON.COM";
};

（可左右滑动）

注意上面标注的Client与KBHttpUtils代码中SpnegoHttpURLConnection spnego = new SpnegoHttpURLConnection("Client")一致。

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

5.在pom文件内容如下

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>cdh-project</artifactId>
        <groupId>com.cloudera</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>livy-demo</artifactId>
    <packaging>jar</packaging>
    <name>livy-demo</name>
    <url>http://maven.apache.org</url>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <repositories>
        <repository>
            <id>maven2</id>
            <url>http://repository.jspresso.org/maven2/</url>
            <name>Maven2 Repositories</name>
            <snapshots>
                <enabled>false</enabled>
            </snapshots>
        </repository>
    </repositories>
    <dependencies>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.5.4</version>
        </dependency>
        <dependency>
            <groupId>net.sourceforge.spnego</groupId>
            <artifactId>spnego</artifactId>
            <version>7.0</version>
        </dependency>
    </dependencies>
</project>

（可左右滑动）

3.编写示例代码

1.Kerberos的HTTP请求工具类（KBHttpUtils.java）

package com.cloudera.utils;
import net.sourceforge.spnego.SpnegoHttpURLConnection;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import java.util.Map;
/**
 * package: com.cloudera.utils
 * describe: 访问Kerberos环境的Http工具类
 * creat_user: Fayson
 * email: htechinfo@163.com
 * creat_date: 2018/2/12
 * creat_time: 下午4:57
 * 公众号：Hadoop实操
 */
public class KBHttpUtils {
    /**
     * HttpGET请求
     * @param url
     * @param headers
     * @return
     */
    public static String getAccess(String url, Map<String,String> headers) {
        StringBuilder sb = new StringBuilder();
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        InputStream in = null;
        try {
            final SpnegoHttpURLConnection spnego = new SpnegoHttpURLConnection("Client");
            spnego.setRequestMethod("GET");
            if(headers != null && headers.size() > 0){
                headers.forEach((K,V)->spnego.setRequestProperty(K,V));
            }
            spnego.connect(new URL(url),bos);
            in = spnego.getInputStream();
            byte[] b = new byte[1024];
            int len ;
            while ((len = in.read(b)) > 0) {
                sb.append(new String(b, 0, len));
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (bos != null) {
                try {
                    bos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        System.out.println("Result:" + sb.toString());
        return sb.toString();
    }
    /**
     * HttpDelete请求
     * @param url
     * @param headers
     * @return
     */
    public  static String deleteAccess(String url, Map<String,String> headers) {
        StringBuilder sb = new StringBuilder();
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        InputStream in = null;
        try {
            final SpnegoHttpURLConnection spnego = new SpnegoHttpURLConnection("Client");
            spnego.setRequestMethod("DELETE");
            if(headers != null && headers.size() > 0){
                headers.forEach((K,V)->spnego.setRequestProperty(K,V));
            }
            spnego.connect(new URL(url),bos);
            in = spnego.getInputStream();
            byte[] b = new byte[1024];
            int len ;
            while ((len = in.read(b)) > 0) {
                sb.append(new String(b, 0, len));
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (bos != null) {
                try {
                    bos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        System.out.println("Result:" + sb.toString());
        return sb.toString();
    }
    /**
     * HttpPost请求
     * @param url
     * @param headers
     * @param data
     * @return
     */
    public static String postAccess(String url, Map<String,String> headers, String data)  {
        StringBuilder sb = new StringBuilder();
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        InputStream in = null;
        try {
            final SpnegoHttpURLConnection spnego = new SpnegoHttpURLConnection("Client");
            spnego.setRequestMethod("POST");
            if(headers != null && headers.size() > 0){
                headers.forEach((K,V)->spnego.setRequestProperty(K,V));
            }
            if(data != null){
                bos.write(data.getBytes());
            }
            spnego.connect(new URL(url),bos);
            System.out.println("Kerberos data:"+data);
            System.out.println("HTTP Status Code: " + spnego.getResponseCode());
            System.out.println("HTTP Status Message: "+ spnego.getResponseMessage());
            in = spnego.getInputStream();
            byte[] b = new byte[1024];
            int len ;
            while ((len = in.read(b)) > 0) {
                sb.append(new String(b, 0, len));
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (bos != null) {
                try {
                    bos.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        System.out.println("Result:" + sb.toString());
        return sb.toString();
    }
}

（可左右滑动）

2.Livy RESTful API调用示例代码

package com.cloudera.kerberos;
import com.cloudera.utils.KBHttpUtils;
import java.util.HashMap;
/**
 * package: com.cloudera
 * describe: Kerberos环境下Livy RESTful API接口调用
 * 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-21-83.ap-southeast-1.compute.internal:8998";
    public static void main(String[] args) {
        System.setProperty("java.security.krb5.conf", "/Volumes/Transcend/keytab/krb5.conf");
        System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
//        System.setProperty("sun.security.krb5.debug", "true"); //Kerberos Debug模式
        System.setProperty("java.security.auth.login.config", "/Volumes/Transcend/keytab/login-yarn.conf");
        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\"}";
//        KBHttpUtils.postAccess(LIVY_HOST + "/sessions", headers, kindJson);
        //执行code
        String code = "{\"code\":\"sc.parallelize(1 to 2).count()\"}";
//        KBHttpUtils.postAccess(LIVY_HOST + "/sessions/2/statements", headers, code);
        //删除会话
//        KBHttpUtils.deleteAccess(LIVY_HOST + "/sessions/3", 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.14.0-hadoop2.6.0-cdh5.14.0.jar\"}";
        //向集群提交Spark作业
        KBHttpUtils.postAccess(LIVY_HOST + "/batches", headers, submitJob);
        //通过提交作业返回的SessionID获取具体作业的执行状态及APPID
//        KBHttpUtils.getAccess(LIVY_HOST + "/batches/4", headers);
    }
}

（可左右滑动）

4.示例代码运行

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

响应结果：

{
  "id": 14,
  "state": "starting",
  "appId": null,
  "appInfo": {
    "driverLogUrl": null,
    "sparkUiUrl": null
  },
  "log": ["stdout: ", "\nstderr: ", "\nYARN Diagnostics: "]
}

（可左右滑动）

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

响应结果：

{
  "id": 14,
  "state": "success",
  "appId": "application_1518423607906_0006",
  "appInfo": {
    "driverLogUrl": null,
    "sparkUiUrl": "http://ip-172-31-16-68.ap-southeast-1.compute.internal:8088/proxy/application_1518423607906_0006/"
  },
  "log": ["\t ApplicationMaster RPC port: -1", "\t queue: root.users.fayson", "\t start time: 1518428233280", "\t final status: UNDEFINED", "\t tracking URL: http://ip-172-31-16-68.ap-southeast-1.compute.internal:8088/proxy/application_1518423607906_0006/", "\t user: fayson", "18/02/12 04:37:13 INFO util.ShutdownHookManager: Shutdown hook called", "18/02/12 04:37:13 INFO util.ShutdownHookManager: Deleting directory /tmp/spark-6a934286-5db7-4c6a-aa72-eefc294086a4", "\nstderr: ", "\nYARN Diagnostics: "]
}

（可左右滑动）

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

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

CM上Yarn的应用程序界面显示

Yarn的8088界面显示

Spark的History界面显示

显示作业运行成功

5.总结

在Java 访问Kerberos环境的Livy API接口时需要在代码中加载krb5.conf、login-yran.conf配置文件到环境变量中，实现fayosn@FAYSON.COM用户登录
访问Kerberos环境下的RESTfulAPI接口需要使用HttpClient提供的spnego方式访问，这里Fayson使用第三方封装好的spnego包

GitHub地址：

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

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