【发布时间】：2016-09-03 03:11:57
【问题描述】：
我正在为特定领域的语言开发自定义解释器。根据 Apache Zeppelin 文档 (https://zeppelin.incubator.apache.org/docs/latest/development/writingzeppelininterpreter.html) 中给出的示例，解释器工作得很好。现在我想将一些结果存储在一个新的 DataFrame 中。
我找到了创建 DataFrames 的代码 (http://spark.apache.org/docs/latest/sql-programming-guide.html)，但我无法在我的解释器中使用它，因为我基本上找不到从我的自定义解释器中访问有效运行时 SparkContext（通常称为“sc”）的方法。
我尝试了（静态）SparkContext.getOrCreate()，但这甚至导致了 ClassNotFoundException。然后我将整个 zeppelin-spark-dependencies...jar 添加到我的解释器文件夹中，这解决了类加载问题，但现在我得到了 SparkException（“必须设置主 url...”）。
知道如何从自定义解释器中访问我的 Notebook 的 SparkContext 吗？非常感谢！
更新
感谢 Kangrok Lee 在下面的评论，我的代码现在如下所示：见下文。它运行并且似乎创建了一个 DataFrame（至少它不再抛出任何异常）。但是我不能在随后的 SQL 段落中使用创建的 DataFrame（第一段使用我的“%opl”解释器，如下所示，它应该创建“结果”DataFrame）：
%opl
1 2 3
> 1
> 2
> 3
%sql
select * from result
> Table not found: result; line 1 pos 14
所以我处理 SparkContext 的方式可能仍然存在问题。有任何想法吗？非常感谢！
package opl;
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class OplInterpreter2 extends Interpreter {
static {
Interpreter.register("opl","opl",OplInterpreter2.class.getName(),
new InterpreterPropertyBuilder()
.add("spark.master", "local[4]", "spark.master")
.add("spark.app.name", "Opl Interpreter", "spark.app.name")
.add("spark.serializer", "org.apache.spark.serializer.KryoSerializer", "spark.serializer")
.build());
}
private Logger logger = LoggerFactory.getLogger(OplInterpreter2.class);
private void log(Object o) {
if (logger != null)
logger.warn("OplInterpreter2 "+o);
}
public OplInterpreter2(Properties properties) {
super(properties);
log("CONSTRUCTOR");
}
@Override
public void open() {
log("open()");
}
@Override
public void cancel(InterpreterContext arg0) {
log("cancel()");
}
@Override
public void close() {
log("close()");
}
@Override
public List<String> completion(String arg0, int arg1) {
log("completion()");
return new ArrayList<String>();
}
@Override
public FormType getFormType() {
log("getFormType()");
return FormType.SIMPLE;
}
@Override
public int getProgress(InterpreterContext arg0) {
log("getProgress()");
return 100;
}
@Override
public InterpreterResult interpret(String string, InterpreterContext context) {
log("interpret() "+string);
PrintStream oldSys = System.out;
try {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
PrintStream ps = new PrintStream(baos);
System.setOut(ps);
execute(string);
System.out.flush();
System.setOut(oldSys);
return new InterpreterResult(
InterpreterResult.Code.SUCCESS,
InterpreterResult.Type.TEXT,
baos.toString());
} catch (Exception ex) {
System.out.flush();
System.setOut(oldSys);
return new InterpreterResult(
InterpreterResult.Code.ERROR,
InterpreterResult.Type.TEXT,
ex.toString());
}
}
private void execute(String code) throws Exception {
SparkContext sc = SparkContext.getOrCreate();
SQLContext sqlc = SQLContext.getOrCreate(sc);
StructType structType = new StructType().add("value",DataTypes.IntegerType);
ArrayList<Row> list = new ArrayList<Row>();
for (String s : code.trim().split("\\s+")) {
int value = Integer.parseInt(s);
System.out.println(value);
list.add(RowFactory.create(value));
}
DataFrame df = sqlc.createDataFrame(list,structType);
df.registerTempTable("result");
}
}
【问题讨论】：
标签： java interpreter apache-zeppelin