import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  tmpDir = new File(System.getProperty("java.io.tmpdir")+"/ZeppelinLTest_"+System.currentTimeMillis());
  tmpDir.mkdirs();
  new File(tmpDir, "conf").mkdirs();

  System.setProperty(ConfVars.ZEPPELIN_HOME.getVarName(), tmpDir.getAbsolutePath());
  System.setProperty(ConfVars.ZEPPELIN_INTERPRETERS.getVarName(), "org.apache.zeppelin.interpreter.mock.MockInterpreter1,org.apache.zeppelin.interpreter.mock.MockInterpreter11,org.apache.zeppelin.interpreter.mock.MockInterpreter2");

  conf = ZeppelinConfiguration.create();

  Interpreter.registeredInterpreters = Collections
      .synchronizedMap(new HashMap<String, Interpreter.RegisteredInterpreter>());
  MockInterpreter1.register("mock1", "group1", "org.apache.zeppelin.interpreter.mock.MockInterpreter1");
  MockInterpreter11.register("mock11", "group1", "org.apache.zeppelin.interpreter.mock.MockInterpreter11");
  MockInterpreter2.register("mock2", "group2", "org.apache.zeppelin.interpreter.mock.MockInterpreter2");

  factory = new InterpreterFactory(conf, new InterpreterOption(false), null);
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
private DepInterpreter getDepInterpreter() {
  InterpreterGroup intpGroup = getInterpreterGroup();
  if (intpGroup == null) return null;
  synchronized (intpGroup) {
    for (Interpreter intp : intpGroup) {
      if (intp.getClassName().equals(DepInterpreter.class.getName())) {
        Interpreter p = intp;
        while (p instanceof WrappedInterpreter) {
          p = ((WrappedInterpreter) p).getInnerInterpreter();
        }
        return (DepInterpreter) p;
      }
    }
  }
  return null;
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
private SparkInterpreter getSparkInterpreter() {
  InterpreterGroup intpGroup = getInterpreterGroup();
  if (intpGroup == null) {
    return null;
  }
  synchronized (intpGroup) {
    for (Interpreter intp : intpGroup){
      if (intp.getClassName().equals(SparkInterpreter.class.getName())) {
        Interpreter p = intp;
        while (p instanceof WrappedInterpreter) {
          p = ((WrappedInterpreter) p).getInnerInterpreter();
        }
        return (SparkInterpreter) p;
      }
    }
  }
  return null;
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Override
public Scheduler getScheduler() {
  if (concurrentSQL()) {
    int maxConcurrency = 10;
    return SchedulerFactory.singleton().createOrGetParallelScheduler(
        SparkSqlInterpreter.class.getName() + this.hashCode(), maxConcurrency);
  } else {
    // getSparkInterpreter() calls open() inside.
    // That means if SparkInterpreter is not opened, it'll wait until SparkInterpreter open.
    // In this moment UI displays 'READY' or 'FINISHED' instead of 'PENDING' or 'RUNNING'.
    // It's because of scheduler is not created yet, and scheduler is created by this function.
    // Therefore, we can still use getSparkInterpreter() here, but it's better and safe
    // to getSparkInterpreter without opening it.
    for (Interpreter intp : getInterpreterGroup()) {
      if (intp.getClassName().equals(SparkInterpreter.class.getName())) {
        Interpreter p = intp;
        return p.getScheduler();
      } else {
        continue;
      }
    }
    throw new InterpreterException("Can't find SparkInterpreter");
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Test
public void testInterpreterUnbindOfNullReplParagraph() throws IOException {
  // create note
  Note note1 = notebook.createNote(anonymous);

  // add paragraph with invalid magic
  Paragraph p1 = note1.addNewParagraph(AuthenticationInfo.ANONYMOUS);
  p1.setText("%fake ");

  // make sure that p1's repl is null
  Interpreter intp = p1.getBindedInterpreter();
  assertEquals(intp, null);

  // Unbind all interpreter from note
  // NullPointerException shouldn't occur here
  notebook.bindInterpretersToNote("user", note1.getId(), new LinkedList<String>());

  // remove note
  notebook.removeNote(note1.getId(), anonymous);
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Test
public void testSingleInterpreterProcess() throws InterpreterException, IOException {
  InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
  interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED);

  Interpreter interpreter1 = interpreterSetting.getDefaultInterpreter("user1", "note1");
  RemoteInterpreter remoteInterpreter1 = (RemoteInterpreter) interpreter1;
  InterpreterContext context1 = new InterpreterContext("noteId", "paragraphId", "repl",
      "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap<String, Object>(), new GUI(),
      new GUI(), null, null, new ArrayList<InterpreterContextRunner>(), null);
  remoteInterpreter1.interpret("hello", context1);

  assertEquals(1, interpreterSettingManager.getRecoveryStorage().restore().size());

  interpreterSetting.close();
  assertEquals(0, interpreterSettingManager.getRecoveryStorage().restore().size());
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
protected Interpreter getInterpreter(String sessionId, String className) throws TException {
  if (interpreterGroup == null) {
    throw new TException(
        new InterpreterException("Interpreter instance " + className + " not created"));
  }
  synchronized (interpreterGroup) {
    List<Interpreter> interpreters = interpreterGroup.get(sessionId);
    if (interpreters == null) {
      throw new TException(
          new InterpreterException("Interpreter " + className + " not initialized"));
    }
    for (Interpreter inp : interpreters) {
      if (inp.getClassName().equals(className)) {
        return inp;
      }
    }
  }
  throw new TException(new InterpreterException("Interpreter instance "
      + className + " not found"));
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Override
public void cancel(String noteId, String className, RemoteInterpreterContext interpreterContext)
    throws TException {
  logger.info("cancel {} {}", className, interpreterContext.getParagraphId());
  Interpreter intp = getInterpreter(noteId, className);
  String jobId = interpreterContext.getParagraphId();
  Job job = intp.getScheduler().removeFromWaitingQueue(jobId);

  if (job != null) {
    job.setStatus(Status.ABORT);
  } else {
    try {
      intp.cancel(convert(interpreterContext, null));
    } catch (InterpreterException e) {
      throw new TException("Fail to cancel", e);
    }
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Override
public int getProgress(String sessionId, String className,
                       RemoteInterpreterContext interpreterContext)
    throws TException {
  Integer manuallyProvidedProgress = progressMap.get(interpreterContext.getParagraphId());
  if (manuallyProvidedProgress != null) {
    return manuallyProvidedProgress;
  } else {
    Interpreter intp = getInterpreter(sessionId, className);
    if (intp == null) {
      throw new TException("No interpreter {} existed for session {}".format(
          className, sessionId));
    }
    try {
      return intp.getProgress(convert(interpreterContext, null));
    } catch (InterpreterException e) {
      throw new TException("Fail to getProgress", e);
    }
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
private SparkInterpreter getSparkInterpreter() {
  InterpreterGroup intpGroup = getInterpreterGroup();
  if (intpGroup == null) {
    return null;
  }

  Interpreter p = getInterpreterInTheSameSessionByClassName(SparkInterpreter.class.getName());
  if (p == null) {
    return null;
  }

  while (p instanceof WrappedInterpreter) {
    p = ((WrappedInterpreter) p).getInnerInterpreter();
  }
  return (SparkInterpreter) p;
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Override
public Scheduler getScheduler() {
  if (concurrentSQL()) {
    int maxConcurrency = 10;
    return SchedulerFactory.singleton().createOrGetParallelScheduler(
        SparkSqlInterpreter.class.getName() + this.hashCode(), maxConcurrency);
  } else {
    // getSparkInterpreter() calls open() inside.
    // That means if SparkInterpreter is not opened, it'll wait until SparkInterpreter open.
    // In this moment UI displays 'READY' or 'FINISHED' instead of 'PENDING' or 'RUNNING'.
    // It's because of scheduler is not created yet, and scheduler is created by this function.
    // Therefore, we can still use getSparkInterpreter() here, but it's better and safe
    // to getSparkInterpreter without opening it.

    Interpreter intp =
        getInterpreterInTheSameSessionByClassName(SparkInterpreter.class.getName());
    if (intp != null) {
      return intp.getScheduler();
    } else {
      return null;
    }
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Before
public void setup() throws InterpreterException {
  Properties p = new Properties();
  p.setProperty("spark.master", "local[4]");
  p.setProperty("master", "local[4]");
  p.setProperty("spark.submit.deployMode", "client");
  p.setProperty("spark.app.name", "Zeppelin Test");
  p.setProperty("zeppelin.spark.useHiveContext", "true");
  p.setProperty("zeppelin.spark.maxResult", "1000");
  p.setProperty("zeppelin.spark.importImplicit", "true");
  p.setProperty("zeppelin.pyspark.python", "python");
  p.setProperty("zeppelin.dep.localrepo", Files.createTempDir().getAbsolutePath());

  intpGroup = new InterpreterGroup();
  intpGroup.put("session_1", new LinkedList<Interpreter>());

  SparkInterpreter sparkInterpreter = new SparkInterpreter(p);
  intpGroup.get("session_1").add(sparkInterpreter);
  sparkInterpreter.setInterpreterGroup(intpGroup);
  sparkInterpreter.open();

  iPySparkInterpreter = new IPySparkInterpreter(p);
  intpGroup.get("session_1").add(iPySparkInterpreter);
  iPySparkInterpreter.setInterpreterGroup(intpGroup);
  iPySparkInterpreter.open();
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Test
@Ignore
//TODO activate test when using Java 8 and C* 3.x
public void should_describe_aggregate() throws Exception {
    //Given
    Properties properties = new Properties();
    properties.setProperty(CASSANDRA_HOSTS, "127.0.0.1");
    properties.setProperty(CASSANDRA_PORT,  "9042");
    Interpreter interpreter = new CassandraInterpreter(properties);
    interpreter.open();

    final String query = "DESCRIBE AGGREGATES;";

    //When
    final InterpreterResult actual = interpreter.interpret(query, intrContext);

    //Then
    assertThat(actual.code()).isEqualTo(Code.SUCCESS);

}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Test
@Ignore
//TODO activate test when using Java 8 and C* 3.x
public void should_describe_materialized_view() throws Exception {
    //Given
    Properties properties = new Properties();
    properties.setProperty(CASSANDRA_HOSTS, "127.0.0.1");
    properties.setProperty(CASSANDRA_PORT,  "9042");
    Interpreter interpreter = new CassandraInterpreter(properties);
    interpreter.open();

    final String query = "DESCRIBE MATERIALIZED VIEWS;";

    //When
    final InterpreterResult actual = interpreter.interpret(query, intrContext);

    //Then
    assertThat(actual.code()).isEqualTo(Code.SUCCESS);
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Before
public void setUp() throws InterpreterException {
  Properties properties = new Properties();
  properties.put("zeppelin.pig.execType", "local");
  properties.put("zeppelin.pig.maxResult", "20");

  pigInterpreter = new PigInterpreter(properties);
  pigQueryInterpreter = new PigQueryInterpreter(properties);
  List<Interpreter> interpreters = new ArrayList();
  interpreters.add(pigInterpreter);
  interpreters.add(pigQueryInterpreter);
  InterpreterGroup group = new InterpreterGroup();
  group.put("note_id", interpreters);
  pigInterpreter.setInterpreterGroup(group);
  pigQueryInterpreter.setInterpreterGroup(group);
  pigInterpreter.open();
  pigQueryInterpreter.open();

  context = new InterpreterContext(null, "paragraph_id", null, null, null, null, null, null, null,
      null, null, null, null);
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
/**
 * Run all paragraphs sequentially.
 *
 * @param jobListener
 */
public void runAll() {
  synchronized (paragraphs) {
    for (Paragraph p : paragraphs) {
      p.setNoteReplLoader(replLoader);
      p.setListener(jobListenerFactory.getParagraphJobListener(this));
      Interpreter intp = replLoader.get(p.getRequiredReplName());
      intp.getScheduler().submit(p);
    }
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
/**
 * Run a single paragraph.
 *
 * @param paragraphId
 */
public void run(String paragraphId) {
  Paragraph p = getParagraph(paragraphId);
  p.setNoteReplLoader(replLoader);
  p.setListener(jobListenerFactory.getParagraphJobListener(this));
  Interpreter intp = replLoader.get(p.getRequiredReplName());
  if (intp == null) {
    throw new InterpreterException("Interpreter " + p.getRequiredReplName() + " not found");
  }
  if (p.getConfig().get("enabled") == null || (Boolean) p.getConfig().get("enabled")) {
    intp.getScheduler().submit(p);
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
private RemoteInterpreterProcess getRemoteInterpreterProcess() {
  if (interpreterGroup.size() == 0) {
    throw new RuntimeException("Can't get remoteInterpreterProcess");
  }
  Interpreter p = interpreterGroup.get(0);
  while (p instanceof WrappedInterpreter) {
    p = ((WrappedInterpreter) p).getInnerInterpreter();
  }

  if (p instanceof RemoteInterpreter) {
    return ((RemoteInterpreter) p).getInterpreterProcess();
  } else {
    throw new RuntimeException("Can't get remoteInterpreterProcess");
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
@Override
public void createInterpreter(String className, Map<String, String> properties)
    throws TException {
  try {
    Class<Interpreter> replClass = (Class<Interpreter>) Object.class.forName(className);
    Properties p = new Properties();
    p.putAll(properties);

    Constructor<Interpreter> constructor =
        replClass.getConstructor(new Class[] {Properties.class});
    Interpreter repl = constructor.newInstance(p);

    ClassLoader cl = ClassLoader.getSystemClassLoader();
    repl.setClassloaderUrls(new URL[]{});

    synchronized (interpreterGroup) {
      interpreterGroup.add(new LazyOpenInterpreter(
          new ClassloaderInterpreter(repl, cl)));
    }

    logger.info("Instantiate interpreter {}", className);
    repl.setInterpreterGroup(interpreterGroup);
  } catch (ClassNotFoundException | NoSuchMethodException | SecurityException
      | InstantiationException | IllegalAccessException
      | IllegalArgumentException | InvocationTargetException e) {
    e.printStackTrace();
    throw new TException(e);
  }
}
 

import org.apache.zeppelin.interpreter.Interpreter; //导入依赖的package包/类
private Interpreter getInterpreter(String className) throws TException {
  synchronized (interpreterGroup) {
    for (Interpreter inp : interpreterGroup) {
      if (inp.getClassName().equals(className)) {
        return inp;
      }
    }
  }
  throw new TException(new InterpreterException("Interpreter instance "
      + className + " not found"));
}