Java 创建与读取Livy Session以及结果

本文主要是介绍Java 创建与读取Livy Session以及结果，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

直接上代码

package net.aas.penrose.deploy.utils;import org.springframework.http.*;
import org.springframework.web.client.RestTemplate;import java.util.Map;public  class LivyServerUtil {private static final String HOST = "http://120.132.101.202";private static final String PORT = "8998";private static final String SESSIONS = "sessions";private static final String STATEMENTS = "statements";private static RestTemplate client = new RestTemplate();private static HttpHeaders headers = new HttpHeaders();private static final String ERROR_GETRESULT = "读取结果时出错";private static final String ERROR_CREATE_SESSION = "创建session时出错";private static final String ERROR_CODE = "返回code不是201 or 201";/*** 向目的URL发送post请求* @param host       SPARK_HOME:port* @param code    excute Code* @return  AdToutiaoJsonTokenData*/public static String sendPost(String host,String port, String code){headers.setContentType(MediaType.APPLICATION_JSON);String sessionId = getSession(host,port);String finalURL = host +":" + port + "/" + SESSIONS +"/" + sessionId +"/"+STATEMENTS;HttpEntity<String> entity = new HttpEntity<>(code,headers);ResponseEntity<Map> response = client.exchange(finalURL, HttpMethod.POST, entity, Map.class);if(response.getStatusCode().toString().equals("201") ||response.getStatusCode().toString().equals("202")){String resultBody = response.getBody().toString();String curentIndex = resultBody.substring(resultBody.indexOf("=")+1,resultBody.indexOf(","));//读取结果return getResult(host,port,sessionId,curentIndex);}return ERROR_CODE;}private static String getResult(String host,String port,String sessionId,String index){int i = 1;while( i< 10000){String finalURL = host+":"+port+"/"+SESSIONS+"/"+sessionId+"/"+STATEMENTS +"/"+ index;ResponseEntity<Map> resp = client.exchange(finalURL, HttpMethod.GET, null, Map.class);if(resp.getStatusCode().toString().equals("201")||resp.getStatusCode().toString().equals("200")){String resultBody = resp.getBody().toString();if(resultBody.toString().contains("state=available")){System.out.println("===============Index : "+i+"====================");return resultBody;}}else{return ERROR_CODE;}i++;}return ERROR_GETRESULT;}private static String getSession(String host,String port){ResponseEntity<Map> resp = client.exchange(host+":"+port+SESSIONS+"/", HttpMethod.GET, null, Map.class);System.out.println(resp.getBody().toString()+"getSession()======");String resultBody = resp.getBody().toString();if(resultBody.contains("[]")){try {if(createSparkSession(HOST,PORT)){ResponseEntity<Map> resp2 = client.exchange(host+":"+port+SESSIONS+"/", HttpMethod.GET, null, Map.class);String resultBody2 = resp2.getBody().toString();String sessionId = resultBody2.substring(resultBody2.indexOf("sessions=[{id=")+14,resultBody2.indexOf(", appId"));return sessionId;}else{System.out.println(ERROR_CREATE_SESSION+"======");}} catch (Exception e) {System.out.println(ERROR_CREATE_SESSION);e.printStackTrace();}}return resultBody.substring(resultBody.indexOf("[{id=")+5,resultBody.indexOf("[{id=")+6);}private static boolean createSparkSession(String host,String port) throws Exception{HttpEntity<String> entity = new HttpEntity<String>("{\"kind\": \"spark\"}",headers);ResponseEntity<Map> resp = client.exchange(host+":"+port+SESSIONS+"/", HttpMethod.POST, entity, Map.class);if(resp.getStatusCode().toString().equals("201")||resp.getStatusCode().toString().equals("200"))return true;return false;}public static void main(String args[]){System.out.println(sendPost(HOST,PORT,"{\"code\":\"100+100\"}"));}}