mirror of
https://github.com/apache/zeppelin
synced 2026-05-24 09:38:26 +00:00
working livy
This commit is contained in:
Prabhjyot Singh
parent
40534974ae
commit
ace28a8000
7 changed files with 469 additions and 280 deletions
|
|
@ -62,6 +62,11 @@
|
|||
<artifactId>httpclient</artifactId>
|
||||
<version>4.3.4</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.google.code.gson</groupId>
|
||||
<artifactId>gson</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
|
|
|
|||
116
livy/src/main/java/org/apache/zeppelin/livy/LivyHelper.java
Normal file
116
livy/src/main/java/org/apache/zeppelin/livy/LivyHelper.java
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.zeppelin.livy;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
import com.google.gson.reflect.TypeToken;
|
||||
import org.apache.zeppelin.interpreter.InterpreterContext;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult.Code;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import static org.apache.zeppelin.livy.RequestHelper.executeHTTP;
|
||||
|
||||
/**
|
||||
* Livy interpreter for Zeppelin.
|
||||
*/
|
||||
public class LivyHelper {
|
||||
Logger LOGGER = LoggerFactory.getLogger(LivyHelper.class);
|
||||
|
||||
|
||||
Gson gson = new Gson();
|
||||
|
||||
protected Integer createSession(String user, String kind) {
|
||||
try {
|
||||
String json = executeHTTP(System.getProperty("zeppelin.livy.url") + "/sessions",
|
||||
"POST",
|
||||
"{\"kind\": \"" + kind + "\", \"proxyUser\": \"" + user + "\"}"
|
||||
);
|
||||
Map jsonMap = (Map<String, Object>) gson.fromJson(json,
|
||||
new TypeToken<Map<String, Object>>() {
|
||||
}.getType());
|
||||
return ((Double) jsonMap.get("id")).intValue();
|
||||
} catch (Exception e) {
|
||||
LOGGER.error("Error getting session for user", e);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
public InterpreterResult interpret(String[] lines, InterpreterContext context,
|
||||
Map<String, Integer> userSessionMap) {
|
||||
synchronized (this) {
|
||||
InterpreterResult res = interpretInput(lines, context, userSessionMap);
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
public InterpreterResult interpretInput(String[] lines, InterpreterContext context,
|
||||
Map<String, Integer> userSessionMap) {
|
||||
String[] linesToRun = new String[lines.length + 1];
|
||||
for (int i = 0; i < lines.length; i++) {
|
||||
linesToRun[i] = lines[i];
|
||||
}
|
||||
linesToRun[lines.length] = "print(\"\")";
|
||||
|
||||
String incomplete = "";
|
||||
Code r = null;
|
||||
|
||||
for (int l = 0; l < linesToRun.length; l++) {
|
||||
String s = linesToRun[l];
|
||||
// check if next line starts with "." (but not ".." or "./") it is treated as an invocation
|
||||
if (l + 1 < linesToRun.length) {
|
||||
String nextLine = linesToRun[l + 1].trim();
|
||||
if (nextLine.startsWith(".") && !nextLine.startsWith("..") && !nextLine.startsWith("./")) {
|
||||
incomplete += s + "\n";
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
return new InterpreterResult(InterpreterResult.Code.SUCCESS,
|
||||
executeCommand(incomplete + s, context, userSessionMap));
|
||||
} catch (Exception e) {
|
||||
return new InterpreterResult(Code.ERROR, e.getMessage());
|
||||
}
|
||||
|
||||
}
|
||||
if (r == Code.INCOMPLETE) {
|
||||
return new InterpreterResult(r, "Incomplete expression");
|
||||
} else {
|
||||
return new InterpreterResult(Code.SUCCESS);
|
||||
}
|
||||
}
|
||||
|
||||
private String executeCommand(String lines, InterpreterContext context,
|
||||
Map<String, Integer> userSessionMap) throws Exception {
|
||||
String json = executeHTTP(System.getProperty("zeppelin.livy.url") + "/sessions/"
|
||||
+ userSessionMap.get(context.getAuthenticationInfo().getUser())
|
||||
+ "/statements",
|
||||
"POST",
|
||||
"{\"code\": \"" + lines + "\" }");
|
||||
Map jsonMap = (Map<String, Object>) gson.fromJson(json,
|
||||
new TypeToken<Map<String, Object>>() {
|
||||
}.getType());
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -1,219 +0,0 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.zeppelin.livy;
|
||||
|
||||
import org.apache.commons.exec.ExecuteWatchdog;
|
||||
import org.apache.commons.exec.Executor;
|
||||
import org.apache.zeppelin.interpreter.Interpreter;
|
||||
import org.apache.zeppelin.interpreter.InterpreterContext;
|
||||
import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult.Code;
|
||||
import org.apache.zeppelin.scheduler.Job;
|
||||
import org.apache.zeppelin.scheduler.Scheduler;
|
||||
import org.apache.zeppelin.scheduler.SchedulerFactory;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
import static org.apache.zeppelin.livy.RequestHelper.executeHTTP;
|
||||
|
||||
/**
|
||||
* Livy interpreter for Zeppelin.
|
||||
*/
|
||||
public class LivyInterpreter extends Interpreter {
|
||||
Logger logger = LoggerFactory.getLogger(LivyInterpreter.class);
|
||||
private static final String EXECUTOR_KEY = "executor";
|
||||
int commandTimeOut = 600000;
|
||||
|
||||
|
||||
static final String DEFAULT_URL = "http://localhost:8998";
|
||||
static final String DEFAULT_USER = "livy";
|
||||
|
||||
private Map<String, Integer> userSessionMap;
|
||||
|
||||
static {
|
||||
Interpreter.register(
|
||||
"livy",
|
||||
"livy",
|
||||
LivyInterpreter.class.getName(),
|
||||
new InterpreterPropertyBuilder()
|
||||
.add("zeppelin.livy.url", DEFAULT_URL, "The URL for Livy Server.")
|
||||
.add("zeppelin.livy.user", DEFAULT_USER, "The livy user")
|
||||
.build()
|
||||
);
|
||||
}
|
||||
|
||||
public LivyInterpreter(Properties property) {
|
||||
super(property);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void open() {
|
||||
userSessionMap = new HashMap<>();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public InterpreterResult interpret(String line, InterpreterContext context) {
|
||||
if (userSessionMap.get(context.getAuthenticationInfo().getUser()) == null) {
|
||||
userSessionMap.put(context.getAuthenticationInfo().getUser(), createSession(context.getAuthenticationInfo().getUser()));
|
||||
}
|
||||
if (line == null || line.trim().length() == 0) {
|
||||
return new InterpreterResult(Code.SUCCESS);
|
||||
}
|
||||
return interpret(line.split("\n"), context);
|
||||
}
|
||||
|
||||
private Integer createSession(String user) {
|
||||
try {
|
||||
String json = executeHTTP(DEFAULT_URL + "/sessions", "POST", "{\"kind\": \"pyspark\", \"proxyUser\": \"" + user + "\"}");
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
public InterpreterResult interpret(String[] lines, InterpreterContext context) {
|
||||
synchronized (this) {
|
||||
InterpreterResult res = interpretInput(lines, context);
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
public InterpreterResult interpretInput(String[] lines, InterpreterContext context) {
|
||||
// logger.debug("Run livy command '" + cmd + "'");
|
||||
// CommandLine cmdLine = CommandLine.parse("bash");
|
||||
// cmdLine.addArgument("-c", false);
|
||||
// cmdLine.addArgument(cmd, false);
|
||||
// DefaultExecutor executor = new DefaultExecutor();
|
||||
// ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
|
||||
// ByteArrayOutputStream errorStream = new ByteArrayOutputStream();
|
||||
// executor.setStreamHandler(new PumpStreamHandler(contextInterpreter.out, errorStream));
|
||||
// executor.setWatchdog(new ExecuteWatchdog(commandTimeOut));
|
||||
//
|
||||
// Job runningJob = getRunningJob(contextInterpreter.getParagraphId());
|
||||
// Map<String, Object> info = runningJob.info();
|
||||
// info.put(EXECUTOR_KEY, executor);
|
||||
// try {
|
||||
// int exitVal = executor.execute(cmdLine);
|
||||
// logger.info("Paragraph " + contextInterpreter.getParagraphId()
|
||||
// + "return with exit value: " + exitVal);
|
||||
// return new InterpreterResult(InterpreterResult.Code.SUCCESS, null);
|
||||
// } catch (ExecuteException e) {
|
||||
// int exitValue = e.getExitValue();
|
||||
// logger.error("Can not run " + cmd, e);
|
||||
// Code code = Code.ERROR;
|
||||
// String msg = errorStream.toString();
|
||||
// if (exitValue == 143) {
|
||||
// code = Code.INCOMPLETE;
|
||||
// msg = msg + "Paragraph received a SIGTERM.\n";
|
||||
// logger.info("The paragraph " + contextInterpreter.getParagraphId()
|
||||
// + " stopped executing: " + msg);
|
||||
// }
|
||||
// msg += "ExitValue: " + exitValue;
|
||||
// return new InterpreterResult(code, msg);
|
||||
// } catch (IOException e) {
|
||||
// logger.error("Can not run " + cmd, e);
|
||||
// return new InterpreterResult(Code.ERROR, e.getMessage());
|
||||
// }
|
||||
|
||||
|
||||
String[] linesToRun = new String[lines.length + 1];
|
||||
for (int i = 0; i < lines.length; i++) {
|
||||
linesToRun[i] = lines[i];
|
||||
}
|
||||
linesToRun[lines.length] = "print(\"\")";
|
||||
|
||||
String incomplete = "";
|
||||
Code r = null;
|
||||
|
||||
for (int l = 0; l < linesToRun.length; l++) {
|
||||
String s = linesToRun[l];
|
||||
// check if next line starts with "." (but not ".." or "./") it is treated as an invocation
|
||||
if (l + 1 < linesToRun.length) {
|
||||
String nextLine = linesToRun[l + 1].trim();
|
||||
if (nextLine.startsWith(".") && !nextLine.startsWith("..") && !nextLine.startsWith("./")) {
|
||||
incomplete += s + "\n";
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
executeHTTP()
|
||||
// incomplete + s;
|
||||
// return new InterpreterResult(Code.ERROR, InterpreterUtils.getMostRelevantMessage(e));
|
||||
}
|
||||
if (r == Code.INCOMPLETE) {
|
||||
return new InterpreterResult(r, "Incomplete expression");
|
||||
} else {
|
||||
return new InterpreterResult(Code.SUCCESS);
|
||||
}
|
||||
}
|
||||
|
||||
private Job getRunningJob(String paragraphId) {
|
||||
Job foundJob = null;
|
||||
Collection<Job> jobsRunning = getScheduler().getJobsRunning();
|
||||
for (Job job : jobsRunning) {
|
||||
if (job.getId().equals(paragraphId)) {
|
||||
foundJob = job;
|
||||
}
|
||||
}
|
||||
return foundJob;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel(InterpreterContext context) {
|
||||
Job runningJob = getRunningJob(context.getParagraphId());
|
||||
if (runningJob != null) {
|
||||
Map<String, Object> info = runningJob.info();
|
||||
Object object = info.get(EXECUTOR_KEY);
|
||||
if (object != null) {
|
||||
Executor executor = (Executor) object;
|
||||
ExecuteWatchdog watchdog = executor.getWatchdog();
|
||||
watchdog.destroyProcess();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public FormType getFormType() {
|
||||
return FormType.SIMPLE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getProgress(InterpreterContext context) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Scheduler getScheduler() {
|
||||
return SchedulerFactory.singleton().createOrGetParallelScheduler(
|
||||
LivyInterpreter.class.getName() + this.hashCode(), 10);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> completion(String buf, int cursor) {
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,145 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.zeppelin.livy;
|
||||
|
||||
import org.apache.commons.exec.DefaultExecutor;
|
||||
import org.apache.commons.exec.ExecuteWatchdog;
|
||||
import org.apache.commons.exec.Executor;
|
||||
import org.apache.commons.exec.PumpStreamHandler;
|
||||
import org.apache.commons.io.output.ByteArrayOutputStream;
|
||||
import org.apache.zeppelin.interpreter.Interpreter;
|
||||
import org.apache.zeppelin.interpreter.InterpreterContext;
|
||||
import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult;
|
||||
import org.apache.zeppelin.scheduler.Job;
|
||||
import org.apache.zeppelin.scheduler.Scheduler;
|
||||
import org.apache.zeppelin.scheduler.SchedulerFactory;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
|
||||
/**
|
||||
* Livy PySpark interpreter for Zeppelin.
|
||||
*/
|
||||
public class LivyPySparkInterpreter extends Interpreter {
|
||||
|
||||
static String DEFAULT_URL = "http://192.168.0.106:8080";
|
||||
|
||||
static {
|
||||
Interpreter.register(
|
||||
"livy",
|
||||
"pyspark",
|
||||
LivyHelper.class.getName(),
|
||||
new InterpreterPropertyBuilder()
|
||||
.add("zeppelin.livy.url", DEFAULT_URL, "The URL for Livy Server.")
|
||||
.build()
|
||||
);
|
||||
}
|
||||
|
||||
private Map<String, Integer> userSessionMap;
|
||||
private LivyHelper livyHelper;
|
||||
private static final String EXECUTOR_KEY = "executor";
|
||||
Logger LOGGER = LoggerFactory.getLogger(LivyPySparkInterpreter.class);
|
||||
|
||||
public LivyPySparkInterpreter(Properties property) {
|
||||
super(property);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void open() {
|
||||
userSessionMap = new HashMap<>();
|
||||
livyHelper = new LivyHelper();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public InterpreterResult interpret(String line, InterpreterContext context) {
|
||||
if (userSessionMap.get(context.getAuthenticationInfo().getUser()) == null) {
|
||||
userSessionMap.put(
|
||||
context.getAuthenticationInfo().getUser(),
|
||||
livyHelper.createSession(context.getAuthenticationInfo().getUser(), "pyspark"));
|
||||
}
|
||||
if (line == null || line.trim().length() == 0) {
|
||||
return new InterpreterResult(InterpreterResult.Code.SUCCESS);
|
||||
}
|
||||
|
||||
DefaultExecutor executor = new DefaultExecutor();
|
||||
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
|
||||
ByteArrayOutputStream errorStream = new ByteArrayOutputStream();
|
||||
executor.setStreamHandler(new PumpStreamHandler(context.out, errorStream));
|
||||
executor.setWatchdog(new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT));
|
||||
|
||||
Job runningJob = getRunningJob(context.getParagraphId());
|
||||
Map<String, Object> info = runningJob.info();
|
||||
info.put(EXECUTOR_KEY, executor);
|
||||
|
||||
return livyHelper.interpret(line.split("\n"), context, userSessionMap);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel(InterpreterContext context) {
|
||||
Job runningJob = getRunningJob(context.getParagraphId());
|
||||
if (runningJob != null) {
|
||||
Map<String, Object> info = runningJob.info();
|
||||
Object object = info.get(EXECUTOR_KEY);
|
||||
if (object != null) {
|
||||
Executor executor = (Executor) object;
|
||||
ExecuteWatchdog watchdog = executor.getWatchdog();
|
||||
watchdog.destroyProcess();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public FormType getFormType() {
|
||||
return FormType.SIMPLE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getProgress(InterpreterContext context) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Scheduler getScheduler() {
|
||||
return SchedulerFactory.singleton().createOrGetParallelScheduler(
|
||||
LivyHelper.class.getName() + this.hashCode(), 10);
|
||||
}
|
||||
|
||||
private Job getRunningJob(String paragraphId) {
|
||||
Job foundJob = null;
|
||||
Collection<Job> jobsRunning = getScheduler().getJobsRunning();
|
||||
for (Job job : jobsRunning) {
|
||||
if (job.getId().equals(paragraphId)) {
|
||||
foundJob = job;
|
||||
}
|
||||
}
|
||||
return foundJob;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> completion(String buf, int cursor) {
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,144 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.zeppelin.livy;
|
||||
|
||||
import org.apache.commons.exec.DefaultExecutor;
|
||||
import org.apache.commons.exec.ExecuteWatchdog;
|
||||
import org.apache.commons.exec.Executor;
|
||||
import org.apache.commons.exec.PumpStreamHandler;
|
||||
import org.apache.commons.io.output.ByteArrayOutputStream;
|
||||
import org.apache.zeppelin.interpreter.Interpreter;
|
||||
import org.apache.zeppelin.interpreter.InterpreterContext;
|
||||
import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder;
|
||||
import org.apache.zeppelin.interpreter.InterpreterResult;
|
||||
import org.apache.zeppelin.scheduler.Job;
|
||||
import org.apache.zeppelin.scheduler.Scheduler;
|
||||
import org.apache.zeppelin.scheduler.SchedulerFactory;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
/**
|
||||
* Livy Spark interpreter for Zeppelin.
|
||||
*/
|
||||
public class LivySparkInterpreter extends Interpreter {
|
||||
|
||||
static {
|
||||
Interpreter.register(
|
||||
"livy",
|
||||
"spark",
|
||||
LivyHelper.class.getName(),
|
||||
new InterpreterPropertyBuilder()
|
||||
.build()
|
||||
);
|
||||
}
|
||||
|
||||
private Map<String, Integer> userSessionMap;
|
||||
private LivyHelper livyHelper;
|
||||
private static final String EXECUTOR_KEY = "executor";
|
||||
Logger LOGGER = LoggerFactory.getLogger(LivySparkInterpreter.class);
|
||||
|
||||
public LivySparkInterpreter(Properties property) {
|
||||
super(property);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void open() {
|
||||
userSessionMap = new HashMap<>();
|
||||
livyHelper = new LivyHelper();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public InterpreterResult interpret(String line, InterpreterContext context) {
|
||||
if (context.getAuthenticationInfo().getUser() == null) {
|
||||
|
||||
}
|
||||
if (userSessionMap.get(context.getAuthenticationInfo().getUser()) == null) {
|
||||
userSessionMap.put(
|
||||
context.getAuthenticationInfo().getUser(),
|
||||
livyHelper.createSession(context.getAuthenticationInfo().getUser(), "spark"));
|
||||
}
|
||||
if (line == null || line.trim().length() == 0) {
|
||||
return new InterpreterResult(InterpreterResult.Code.SUCCESS);
|
||||
}
|
||||
|
||||
DefaultExecutor executor = new DefaultExecutor();
|
||||
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
|
||||
ByteArrayOutputStream errorStream = new ByteArrayOutputStream();
|
||||
executor.setStreamHandler(new PumpStreamHandler(context.out, errorStream));
|
||||
executor.setWatchdog(new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT));
|
||||
|
||||
Job runningJob = getRunningJob(context.getParagraphId());
|
||||
Map<String, Object> info = runningJob.info();
|
||||
info.put(EXECUTOR_KEY, executor);
|
||||
|
||||
return livyHelper.interpret(line.split("\n"), context, userSessionMap);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void cancel(InterpreterContext context) {
|
||||
Job runningJob = getRunningJob(context.getParagraphId());
|
||||
if (runningJob != null) {
|
||||
Map<String, Object> info = runningJob.info();
|
||||
Object object = info.get(EXECUTOR_KEY);
|
||||
if (object != null) {
|
||||
Executor executor = (Executor) object;
|
||||
ExecuteWatchdog watchdog = executor.getWatchdog();
|
||||
watchdog.destroyProcess();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public FormType getFormType() {
|
||||
return FormType.SIMPLE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getProgress(InterpreterContext context) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Scheduler getScheduler() {
|
||||
return SchedulerFactory.singleton().createOrGetParallelScheduler(
|
||||
LivyHelper.class.getName() + this.hashCode(), 10);
|
||||
}
|
||||
|
||||
private Job getRunningJob(String paragraphId) {
|
||||
Job foundJob = null;
|
||||
Collection<Job> jobsRunning = getScheduler().getJobsRunning();
|
||||
for (Job job : jobsRunning) {
|
||||
if (job.getId().equals(paragraphId)) {
|
||||
foundJob = job;
|
||||
}
|
||||
}
|
||||
return foundJob;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> completion(String buf, int cursor) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,20 +1,25 @@
|
|||
package org.apache.zeppelin.livy;
|
||||
|
||||
import com.google.gson.Gson;
|
||||
import org.apache.http.HttpResponse;
|
||||
import org.apache.http.client.HttpClient;
|
||||
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.HttpClientBuilder;
|
||||
import org.apache.http.params.HttpConnectionParams;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
import java.io.InputStreamReader;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Created by prabhjyot.singh on 17/02/16.
|
||||
*/
|
||||
public class RequestHelper {
|
||||
static Logger LOGGER = LoggerFactory.getLogger(RequestHelper.class);
|
||||
|
||||
public static String executeHTTP(String targetURL, String method, String jsonData)
|
||||
throws Exception {
|
||||
|
|
@ -22,6 +27,7 @@ public class RequestHelper {
|
|||
HttpClient client = HttpClientBuilder.create().build();
|
||||
|
||||
HttpResponse response;
|
||||
LOGGER.error(jsonData);
|
||||
if (method.equals("POST")) {
|
||||
HttpPost request = new HttpPost(targetURL);
|
||||
request.addHeader("Content-Type", "application/json");
|
||||
|
|
@ -35,70 +41,61 @@ public class RequestHelper {
|
|||
}
|
||||
|
||||
|
||||
if (response.getStatusLine().getStatusCode() == 200 || response.getStatusLine().getStatusCode() == 201 ) {
|
||||
return null;
|
||||
}
|
||||
BufferedReader rd = new BufferedReader(
|
||||
new InputStreamReader(response.getEntity().getContent()));
|
||||
if (response.getStatusLine().getStatusCode() == 200
|
||||
|| response.getStatusLine().getStatusCode() == 201) {
|
||||
BufferedReader rd = new BufferedReader(
|
||||
new InputStreamReader(response.getEntity().getContent()));
|
||||
|
||||
StringBuffer result = new StringBuffer();
|
||||
String line = "";
|
||||
while ((line = rd.readLine()) != null) {
|
||||
result.append(line);
|
||||
StringBuffer result = new StringBuffer();
|
||||
String line = "";
|
||||
while ((line = rd.readLine()) != null) {
|
||||
result.append(line);
|
||||
}
|
||||
return result.toString();
|
||||
}
|
||||
return result.toString();
|
||||
// HttpURLConnection connection = null;
|
||||
// try {
|
||||
// //Create connection
|
||||
// URL url = new URL(targetURL);
|
||||
// connection = (HttpURLConnection) url.openConnection();
|
||||
// connection.setRequestProperty("Content-Type",
|
||||
// "application/json; charset=UTF-8");
|
||||
// connection.setDoOutput(true);
|
||||
// connection.setDoInput(true);
|
||||
// connection.setRequestMethod(method);
|
||||
//
|
||||
// connection.setRequestProperty("Content-Length",
|
||||
// Integer.toString(urlParameters.getBytes().length));
|
||||
// connection.setRequestProperty("Content-Language", "en-US");
|
||||
//
|
||||
// connection.setUseCaches(false);
|
||||
// connection.setDoOutput(true);
|
||||
//
|
||||
// //Send request
|
||||
// OutputStreamWriter wr = new OutputStreamWriter(connection.getOutputStream());
|
||||
// if (jsonData != null) {
|
||||
// wr.write(jsonData);
|
||||
// }
|
||||
// wr.close();
|
||||
//
|
||||
// //Get Response
|
||||
// InputStream is = connection.getInputStream();
|
||||
// BufferedReader rd = new BufferedReader(new InputStreamReader(is));
|
||||
// StringBuilder response = new StringBuilder(); // or StringBuffer if not Java 5+
|
||||
// String line;
|
||||
// while ((line = rd.readLine()) != null) {
|
||||
// response.append(line);
|
||||
// response.append('\r');
|
||||
// }
|
||||
// rd.close();
|
||||
// return response.toString();
|
||||
// } catch (Exception e) {
|
||||
// e.printStackTrace();
|
||||
// throw e;
|
||||
// } finally {
|
||||
// if (connection != null) {
|
||||
// connection.disconnect();
|
||||
// }
|
||||
// }
|
||||
return null;
|
||||
}
|
||||
|
||||
public static void main(String[] asd) throws Exception {
|
||||
String host = "http://192.168.0.106";
|
||||
String sessionCreate = executeHTTP(host + ":8998/sessions", "POST", "{\"kind\": \"pyspark\", \"proxyUser\": \"bob222\"}");
|
||||
System.out.println(sessionCreate);
|
||||
// String host = "http://192.168.0.106";
|
||||
// String sessionCreate = executeHTTP(host + ":8998/sessions", "POST",
|
||||
// "{\"kind\": \"pyspark\", \"proxyUser\": \"bob222\"}");
|
||||
// System.out.println(sessionCreate);
|
||||
//
|
||||
// String countSession = executeHTTP(host + ":8998/sessions", "GET", null);
|
||||
// System.out.println(countSession);
|
||||
String user = "user";
|
||||
String DEFAULT_URL = "http://192.168.0.106:8998";
|
||||
// String json = executeHTTP(DEFAULT_URL + "/sessions", "POST",
|
||||
// "{\"kind\": \"pyspark\", \"proxyUser\": \"" + user + "\"}");
|
||||
Gson gson = new Gson();
|
||||
// Map jsonMap = (Map<String, Object>) gson.fromJson(json,
|
||||
// new TypeToken<Map<String, Object>>() {}.getType());
|
||||
|
||||
String countSession = executeHTTP(host + ":8998/sessions", "GET", null);
|
||||
System.out.println(countSession);
|
||||
Map<String, Integer> stringIntegerHashMap = new HashMap<>();
|
||||
stringIntegerHashMap.put(null, 1);
|
||||
System.out.println();
|
||||
// String lines = "import time\\n" +
|
||||
// "REFRAIN = '''\\n" +
|
||||
// "%d bottles of beer on the wall,\\n" +
|
||||
// "%d bottles of beer,\\n" +
|
||||
// "take one down, pass it around,\\n" +
|
||||
// "%d bottles of beer on the wall!\\n" +
|
||||
// "'''\\n" +
|
||||
// "bottles_of_beer = 10\\n" +
|
||||
// "while bottles_of_beer > 1:\\n" +
|
||||
// " time.sleep(1)\\n" +
|
||||
// " print REFRAIN % (bottles_of_beer, bottles_of_beer,\\n" +
|
||||
// " bottles_of_beer - 1)\\n" +
|
||||
// " bottles_of_beer -= 1\\n";
|
||||
// String json = executeHTTP(DEFAULT_URL + "/sessions/"
|
||||
// + 0
|
||||
// + "/statements",
|
||||
// "POST",
|
||||
// "{\"code\": \"" + lines + "\" }");
|
||||
// Map jsonMap = (Map<String, Object>) gson.fromJson(json,
|
||||
// new TypeToken<Map<String, Object>>() {
|
||||
// }.getType());
|
||||
// System.out.println(json);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -444,7 +444,8 @@ public class ZeppelinConfiguration extends XMLConfiguration {
|
|||
+ "org.apache.zeppelin.markdown.Markdown,"
|
||||
+ "org.apache.zeppelin.angular.AngularInterpreter,"
|
||||
+ "org.apache.zeppelin.shell.ShellInterpreter,"
|
||||
+ "org.apache.zeppelin.livy.LivyInterpreter,"
|
||||
+ "org.apache.zeppelin.livy.LivyPySparkInterpreter,"
|
||||
+ "org.apache.zeppelin.livy.LivySparkInterpreter,"
|
||||
+ "org.apache.zeppelin.hive.HiveInterpreter,"
|
||||
+ "org.apache.zeppelin.tachyon.TachyonInterpreter,"
|
||||
+ "org.apache.zeppelin.phoenix.PhoenixInterpreter,"
|
||||
|
|
|
|||
Loading…
Reference in a new issue