如何使用Java调用CM的API动态配置Yarn资源池

2018-08-17 17:28:12 浏览数 (1)

温馨提示:如果使用电脑查看图片不清晰,可以使用手机打开文章单击文中的图片放大查看高清原图。

Fayson的github: https://github.com/fayson/cdhproject

提示:代码块部分可以左右滑动查看噢

1.文档编写目的


用户在使用CDH集群大数据平台时会有需求在自己的统一管理平台上通过API接口能够动态的设置Yarn资源池,Cloudera Manager提供了丰富的API接口对CDH集群的各个服务进行配置等操作,在上一篇文章Fayson介绍了《如何使用curl命令调用CM的API动态配置Yarn资源池》,本篇文章Fayson主要介绍如何使用Java调用CM的API接口动态的配置Yarn资源池并使其生效。

  • 本次Fayson的测试环境

1.CM和CDH版本为5.15

2.获取Cloudera Manager的API接口


1.登录Cloudera Manager,点击菜单“支持”->“API文档”

2.进入API文档界面,可以看到API的版本为v19

3.本篇文章要用到的API接口说明

代码语言:javascript复制
http://<cm_server>:7180/clusters
http://<cm_server>:7180/api/v19/clusters/{clusterName}/services/{serviceName}/config
http://<cm_server>:7180/api/v19/clusters/{clusterName}/commands/poolsRefresh

(可左右滑动)

共使用到3个接口:

第一个接口用于获取集群的信息,如集群的名称,以供2、3接口使用

第二个接口用于设置Yarn的资源池,接口中的{serviceName}修改为yarn

第三个接口用于刷新Yarn的资源池

3.Java示例工程


1.使用Maven创建cmapi-demo的Java工程,pom.xml内容如下:

代码语言:javascript复制
<dependencies>
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.4</version>
    </dependency>
    <dependency>
        <groupId>net.sf.json-lib</groupId>
        <artifactId>json-lib</artifactId>
        <version>2.4</version>
        <classifier>jdk15</classifier>
    </dependency>
</dependencies>

(可左右滑动)

2.创建HttpUtils.java工具类,该类主要使用HttpClient方式调用CM的API接口

代码语言:javascript复制
package com.cloudera.utils;

import org.apache.commons.lang.StringEscapeUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
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.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.BasicCredentialsProvider;
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.Map;

/**
 * package: com.cloudera
 * describe: Http请求工具类
 * creat_user: Fayson
 * email: htechinfo@163.com
 * creat_date: 2018/2/12
 * creat_time: 下午12:16
 * 公众号:Hadoop实操
 */
public class HttpUtils {

    /**
     * Get方式用户名和密码认证
     * @param url
     * @param headers
     * @param username
     * @param password
     * @return
     */
    public static String getAccessByAuth(String url, Map<String, String> headers, String username, String password) {
        String result = null;

        URI uri = URI.create(url);
        CredentialsProvider credsProvider = new BasicCredentialsProvider();

        credsProvider.setCredentials(new AuthScope(uri.getHost(), uri.getPort()),
                new UsernamePasswordCredentials(username, password));

        CloseableHttpClient httpClient = HttpClients.custom().setDefaultCredentialsProvider(credsProvider)
                .build();

        HttpGet httpGet = new HttpGet(uri);
        if(headers != null && headers.size() > 0){
            headers.forEach((K,V)->httpGet.addHeader(K,V));
        }

        HttpResponse response = null;
        try {
            response = httpClient.execute(httpGet);
            HttpEntity resultEntity = response.getEntity();
            result = EntityUtils.toString(resultEntity);

            return result;
        } catch (IOException e) {
            e.printStackTrace();
        }


        return null;
    }

    /**
     * Post方式用户名和密码认证
     * @param url
     * @param headers
     * @param data
     * @param username
     * @param password
     * @return
     */
    public static  String postAccessByAuth(String url, Map<String, String> headers, String data, String username, String password) {
        String result = null;

        URI uri = URI.create(url);
        CredentialsProvider credsProvider = new BasicCredentialsProvider();

        credsProvider.setCredentials(new AuthScope(uri.getHost(), uri.getPort()),
                new UsernamePasswordCredentials(username, password));

        CloseableHttpClient httpClient = HttpClients.custom()
                .setDefaultCredentialsProvider(credsProvider)
                .build();

        HttpPost post = new HttpPost(uri);

        if(headers != null && headers.size() > 0){
            headers.forEach((K,V)->post.addHeader(K,V));
        }

        try {
            if(data != null) {
                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);

            return result;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result;
    }

    /**
     * Put方式用户名和密码认证方式
     * @param url
     * @param headers
     * @param data
     * @param username
     * @param password
     * @return
     */
    public static  String putAccessByAuth(String url, Map<String, String> headers, String data, String username, String password) {
        String result = null;

        URI uri = URI.create(url);
        CredentialsProvider credsProvider = new BasicCredentialsProvider();

        credsProvider.setCredentials(new AuthScope(uri.getHost(), uri.getPort()),
                new UsernamePasswordCredentials(username, password));

        CloseableHttpClient httpClient = HttpClients.custom()
                .setDefaultCredentialsProvider(credsProvider)
                .build();

        HttpPut put = new HttpPut(uri);

        if(headers != null && headers.size() > 0){
            headers.forEach((K,V)->put.addHeader(K,V));
        }

        try {
            if(data != null) {
                StringEntity entity = new StringEntity(data);
                entity.setContentEncoding("UTF-8");
                entity.setContentType("application/json");
                put.setEntity(entity);
            }

            HttpResponse response = httpClient.execute(put);
            HttpEntity resultEntity = response.getEntity();
            result = EntityUtils.toString(resultEntity);

            return result;
        } catch (Exception e) {
            e.printStackTrace();
        }

        return result;
    }

}

(可左右滑动)

3.创建RestApiConfPool.java类,调用CM API接口设置资源池

代码语言:javascript复制
package com.cloudera;

import com.cloudera.api.model.ApiConfig;
import com.cloudera.utils.HttpUtils;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringEscapeUtils;

import java.util.HashMap;

/**
 * package: com.cloudera
 * describe: 通过CM提供的API配置Yarn动态资源池
 * creat_user: Fayson
 * email: htechinfo@163.com
 * creat_date: 2018/8/7
 * creat_time: 下午3:14
 * 公众号:Hadoop实操
 */
public class RestApiConfPool {

    private static String REQ_CLUSTER_URL = "http://cdh01.fayson.com:7180/api/v19/clusters";
    private static String REQ_SETPOOL_URL = "http://cdh01.fayson.com:7180/api/v19/clusters/cluster/services/yarn/config";
    private static String REQ_FRESHPOOL_URL = "http://cdh01.fayson.com:7180/api/v19/clusters/cluster/commands/poolsRefresh";
    private static String USERNAME = "admin";
    private static String PASSWORD = "admin";

    public static void main(String[] args) {

        HashMap<String, String> headers = new HashMap<>();
        headers.put("Content-Type", "application/json");
        headers.put("Accept", "application/json");

        //获取CM管理的Cluster集群名称
        String result = HttpUtils.getAccessByAuth(REQ_CLUSTER_URL, headers, USERNAME, PASSWORD);
        System.out.println("获取集群信息:"   result);

        //定义资源池配置的JSON配置
        JSONObject requestJson = new JSONObject();
        JSONArray jsonArray = new JSONArray();
        JSONObject yarnjson = new JSONObject();
        String pool_conf = "{\"defaultFairSharePreemptionThreshold\":null,\"defaultFairSharePreemptionTimeout\":null,\"defaultMinSharePreemptionTimeout\":null,\"defaultQueueSchedulingPolicy\":\"fair\",\"queueMaxAMShareDefault\":null,\"queueMaxAppsDefault\":null,\"queuePlacementRules\":[{\"create\":true,\"name\":\"specified\",\"queue\":null,\"rules\":null},{\"create\":null,\"name\":\"nestedUserQueue\",\"queue\":null,\"rules\":[{\"create\":true,\"name\":\"default\",\"queue\":\"users\",\"rules\":null}]},{\"create\":null,\"name\":\"default\",\"queue\":null,\"rules\":null}],\"queues\":[{\"aclAdministerApps\":\"*\",\"aclSubmitApps\":\"*\",\"allowPreemptionFrom\":null,\"fairSharePreemptionThreshold\":null,\"fairSharePreemptionTimeout\":null,\"minSharePreemptionTimeout\":null,\"name\":\"root\",\"queues\":[{\"aclAdministerApps\":null,\"aclSubmitApps\":null,\"allowPreemptionFrom\":null,\"fairSharePreemptionThreshold\":null,\"fairSharePreemptionTimeout\":null,\"minSharePreemptionTimeout\":null,\"name\":\"default\",\"queues\":[],\"schedulablePropertiesList\":[{\"impalaDefaultQueryMemLimit\":null,\"impalaDefaultQueryOptions\":null,\"impalaMaxMemory\":null,\"impalaMaxQueuedQueries\":null,\"impalaMaxRunningQueries\":null,\"impalaQueueTimeout\":null,\"maxAMShare\":null,\"maxChildResources\":null,\"maxResources\":null,\"maxRunningApps\":null,\"minResources\":null,\"scheduleName\":\"default\",\"weight\":1.0}],\"schedulingPolicy\":\"drf\",\"type\":null},{\"aclAdministerApps\":null,\"aclSubmitApps\":null,\"allowPreemptionFrom\":null,\"fairSharePreemptionThreshold\":null,\"fairSharePreemptionTimeout\":null,\"minSharePreemptionTimeout\":null,\"name\":\"users\",\"queues\":[],\"schedulablePropertiesList\":[{\"impalaDefaultQueryMemLimit\":null,\"impalaDefaultQueryOptions\":null,\"impalaMaxMemory\":null,\"impalaMaxQueuedQueries\":null,\"impalaMaxRunningQueries\":null,\"impalaQueueTimeout\":null,\"maxAMShare\":null,\"maxChildResources\":null,\"maxResources\":null,\"maxRunningApps\":null,\"minResources\":null,\"scheduleName\":\"default\",\"weight\":2.0}],\"schedulingPolicy\":\"drf\",\"type\":\"parent\"}],\"schedulablePropertiesList\":[{\"impalaDefaultQueryMemLimit\":null,\"impalaDefaultQueryOptions\":null,\"impalaMaxMemory\":null,\"impalaMaxQueuedQueries\":null,\"impalaMaxRunningQueries\":null,\"impalaQueueTimeout\":null,\"maxAMShare\":null,\"maxChildResources\":null,\"maxResources\":null,\"maxRunningApps\":null,\"minResources\":null,\"scheduleName\":\"default\",\"weight\":1.0}],\"schedulingPolicy\":\"drf\",\"type\":null}],\"userMaxAppsDefault\":null,\"users\":[]}";
        yarnjson.put("name", "yarn_fs_scheduled_allocations");
        yarnjson.put("value", pool_conf);
        jsonArray.add(yarnjson);
        requestJson.put("items", jsonArray);
        //注意使用PUT提交,否则会请求失败
        result = HttpUtils.putAccessByAuth(REQ_SETPOOL_URL, headers, StringEscapeUtils.unescapeJava(requestJson.toString()), USERNAME, PASSWORD);

        System.out.println("动态设置Yarn资源池:"  result);

        //刷新资源池
        result = HttpUtils.postAccessByAuth(REQ_FRESHPOOL_URL, headers, null, USERNAME, PASSWORD);

        System.out.println("刷新资源池配置:"   result);
    }
}

(可左右滑动)

4.示例运行


1.设置资源池前Yarn的配置如下

2.在RestApiConfPool.java类中运行,如下图显示运行成功

3.登陆Cloudera Manager查看资源池配置,已更新为设置的值且已生效

5.总结


1.通过API接口动态的配置Yarn的资源池,首先要获取Yarn服务所在集群名称,通过接口进行动态设置,设置完成后需要调用刷新的API接口使其生效。

2.在设置Yarn资源池时调用的API接口为PUT方式提交,如果指定其他方式会导致设置失败。

3.设置资源池的JSON字符串,需要注意value跟的数据为转义的字符串,必须带“”,否则会设置失败。

4.在Java 代码中字符串“”会被转义”\”,在设置时需要使用Java工具类提供更的方法StringEscapeUtils.unescapeJava()将转义字符串去掉。

GitHub地址:

https://github.com/fayson/cdhproject/blob/master/cmapidemo/src/main/java/com/cloudera/RestApiConfPool.java

https://github.com/fayson/cdhproject/tree/master/cmapidemo/src/main/java/com/cloudera/utils

提示:代码块部分可以左右滑动查看噢

为天地立心,为生民立命,为往圣继绝学,为万世开太平。 温馨提示:如果使用电脑查看图片不清晰,可以使用手机打开文章单击文中的图片放大查看高清原图。

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

0 人点赞