zeppelin源码分析(3)——主要的class分析(中)

接上篇，本文接着分析zeppelin类图中右上角剩余的类，同样，在分析的过程中，我们重点关注该class的职责划分，以及与其他类配合，完成zeppelin的设计目标的过程。

InterpreterInfoSaving

InterpreterInfoSaving是一个convenient类(提供的功能可以由其他类组合完成，设计目的是为了方便调用)，定义其目的就是为了将原来分散存储在各个interpreter子文件夹中的InterpreterSetting、intepreterBindings和interpreterRepositories集中到一处，方便进行查找和持久化。

原来Interpter的注册方式是static注册方式：即每个Interpreter的实现类都定义一段static初始化段，在该类被第一次加载的时候，将其注册到zeppelin中。例如：下面是Python interpreter的注册代码。

  static {
    Interpreter.register(
"python",
"python",
        PythonInterpreter.class.getName(),
new InterpreterPropertyBuilder()
            .add(ZEPPELIN_PYTHON, DEFAULT_ZEPPELIN_PYTHON,
"Python directory. Default : python (assume python is in your $PATH)")
            .build()
    );
  }

现在这种方式已经Deprecated了，新的Interpreter采用了json配置文件的方式来保存这些配置，并且由zeppelin加载时来解析这些配置，该文件约定为interpreter-setting.json，该文件可以放到如下2种位置：
1. 作为main resources打包到interpreter的jar包中
2. 将其放到${ZEPPELIN_HOME}/interpreter/{interpreter}/interpreter-setting.json，位置下
interpreter-settings.json就近与interpreter的jar包存储在一起，符合封装的原则，但是zeppelin需要知道全局已经注册了哪些interpreter，以及这些interpreter的配置
InterpreterInfoSaving类的唯一实例会被持久化到${ZEPPELIN_HOME}/conf/interpreter.json文件，如下图：

可以看出，InterpreterInfoSaving类是${ZEPPELIN_HOME}/conf/interpreter.json持久化文件相对应的内存对象。

InterpreterSetting

InterpreterSettings是一份Interpreter创建的”contract”，其职责如下：
1. 维护了interpreter相关的元信息，这些信息对于正确运行intepreter进程至关重要，如：与外界环境进行交互（如spark on Yarn集群）、第三方依赖加载等
2. 维护了note和intepreterGroup之间的关系

上篇中我们提到InterpreterGroup是创建jvm、note bind的最小单位，因此，也就决定了在多个note时，创建jvm的个数以及在多个note之间进行变量共享的程度。每个interpreter进程可以有一个或者多个interpreterGroup，每个interpreter实例都从属于其中某一个InterpreterGroup。三者之间的关系如下图：

如下是spark interpreter的settings：

    "2C6QR3FVF": {
"id": "2C6QR3FVF",
"name": "spark",
"group": "spark",
"properties": {
"spark.executor.memory": "",
"args": "",
"zeppelin.spark.printREPLOutput": "true",
"spark.cores.max": "",
"zeppelin.dep.additionalRemoteRepository": "spark-packages,http://dl.bintray.com/spark-packages/maven,false;",
"zeppelin.spark.importImplicit": "true",
"zeppelin.spark.sql.stacktrace": "false",
"zeppelin.spark.concurrentSQL": "false",
"zeppelin.spark.useHiveContext": "true",
"zeppelin.pyspark.python": "python",
"zeppelin.dep.localrepo": "local-repo",
"zeppelin.interpreter.localRepo": "..//local-repo/2C6QR3FVF",
"zeppelin.R.knitr": "true",
"zeppelin.spark.maxResult": "1000",
"master": "local[*]",
"spark.app.name": "Zeppelin",
"zeppelin.R.image.width": "100%",
"zeppelin.R.render.options": "out.format \u003d \u0027html\u0027, comment \u003d NA, echo \u003d FALSE, results \u003d \u0027asis\u0027, message \u003d F, warning \u003d F",
"zeppelin.R.cmd": "R"
      },
"interpreterGroup": [
        {
"class": "org.apache.zeppelin.spark.SparkInterpreter",
"name": "spark"
        },
        {
"class": "org.apache.zeppelin.spark.SparkSqlInterpreter",
"name": "sql"
        },
        {
"class": "org.apache.zeppelin.spark.DepInterpreter",
"name": "dep"
        },
        {
"class": "org.apache.zeppelin.spark.PySparkInterpreter",
"name": "pyspark"
        },
        {
"class": "org.apache.zeppelin.spark.SparkRInterpreter",
"name": "r"
        }
      ],
"dependencies": [],
"option": {
"remote": true,
"perNoteSession": false,
"perNoteProcess": false,
"isExistingProcess": false,
"port": "-1"
      }
    }

InterpreterSettings实例被InterpreterFactory对象初始化并填充。
多个note时，究竟是为每个note创建不同的interpreter进程还是共享同一个interpreter进程，zeppelin UI上提供了三种不同的模式可供选择：

Shared模式

一种解释器只有一个Interpreter进程，并且该进程中只有一个InterpreterGroup，所有的Interpreter实例都从属于该InterpreterGroup，当然，也肯定在同一个进程内部。多个note之间，可以很容易的共享变量。
以下关于3种模式的解释来自于Lee Moon Soo相关文章

Scoped模式

一种repl解释器只有一个Interpreter进程，但是与Shared模式不同，会创建过个InterpreterGroup，每个note关联一个InterpreterGroup。这样每个note相当于有了自己的session，session与session相互隔离，但是仍然由于这些InterpreterGroup仍然在同一个进程中，仍然可以在它们之间共享变量。

Isolated模式

独占式，为每个note创建一个独立的intepreter进程，该进程中创建一个InterpreterGroup实例，为该note的服务的Interpreter实例从属于该InterpreterGroup。

SparkInterpreter的Shared/Scoped/Isolated模式

以SparkInterpreter为例，说明这三种模式对SparkContext和Scala repl共享方式上的差别：

SparkInterpreter Shared

所有的note共享同一个SparkContext和Scala REPL实例，因此，如果其中一个note定义了变量a，另外一个note可以访问并且修改该变量a。

SparkInterpreter Scoped

所有的note共享同一个SparkContext，所有的spark job都是通过同一个SparkContext提交的，但是不同的Scala repl解释器，由于不同享Scala repl，故不存在一个note访问并修改了另一个note定义的变量的问题。

SparkInterpreter Isolated

独占式，每个note都有自己的SparkContext和Scala repl，不共享。

其实这三种模式，底层都是通过InterpreterOption类来控制的。

InterpreterOption

虽然InterpreterOption是一个简单的POJO，但是其字段取值，直接决定了zeppelin在创建interpreter进程时的处理方式，直接体现了”性能和资源占用之间进行trade-off”多种策略：

property	含义	说明
existingProcess	是否连接已有intereter进程	intereter进程可以独立启动，如在远程(不同与zeppelin运行的其他host）节点上，可以采用该手段将repl进程进行分布式部署，并且先于zeppelin启动，让zeppelin连接到这些已有的线程，以解决intereter进程横向扩展的问题。该模式需要显式指定host和port，表示zeppelin主进程(ZeppelinServer所在进程）与该interpreter进程进行的IPC时，通信的socket。
perNoteProcess	每个Note创建一个IntereterGroup进程	这是最细粒度的interpreter进程创建方式，同时也是最粗放的资源使用的方式，如果interpreter都在运行zeppelin的host上启动的话， note数量很多的话，很可能造该host内存耗尽
perNoteSession	每个Note的在同一个IntereterGroup进程中创建不同的Interpreter实例	创建interpreter进程时只创建一个，但是在该进程内部，创建不同的interpreter实例。这种方式比perNoteProcess要节省内存资源。

InterpreterSettings是InterpreterOption的使用方，重要的地方见其getInterpreterGroup和getInterpreterProcessKey方法：

  private String getInterpreterProcessKey(String noteId) {
if (getOption().isExistingProcess) {
return Constants.EXISTING_PROCESS;//existingProcess模式，共享现有的IntereterGroup进程
    } else if (getOption().isPerNoteProcess()) {
return noteId;//perNoteProcess模式，每个note都会创建一个新的进程
    } else {
return SHARED_PROCESS;//sharedProcess模式，所有note共享同一个interpreter进程
    }
  }

//为note创建新的InterpreterGroup或者是关联现有的InterpreterGroup
public InterpreterGroup getInterpreterGroup(String noteId) {
    String key = getInterpreterProcessKey(noteId);
synchronized (interpreterGroupRef) {
if (!interpreterGroupRef.containsKey(key)) {
        String interpreterGroupId = id() + ":" + key;
        InterpreterGroup intpGroup =
            interpreterGroupFactory.createInterpreterGroup(interpreterGroupId, getOption());
        interpreterGroupRef.put(key, intpGroup);
      }
return interpreterGroupRef.get(key);
    }
  }

InterpreterFactory

InterpreterFactory是InterpreterGroupFactory的实现类，承担如下职责：
1. interpreter实例的实际创建者
2. interpreter配置文件的加载与持久化，如loadFromFile()和saveToFile()
3. interpreterSettings的管理
4. 第三方依赖加载器
5. 在zeppelin主进程中建立远程AngularObjectRegistry的本地Proxy——RemoteAngularObjectRegistry，以保证远程interpreter进程与前端angular对象双向绑定。

InterpreterFactory实际执行创建interpreter实例的方式为createInterpretersForNote，具体实现如下：

  public void createInterpretersForNote(
      InterpreterSetting interpreterSetting,
      String noteId,
      String key) {
    InterpreterGroup interpreterGroup = interpreterSetting.getInterpreterGroup(noteId);//调用interpreterSetting.getInterpreterGroup以确定创建InterpreterGroup的策略
    String groupName = interpreterSetting.getGroup();
    InterpreterOption option = interpreterSetting.getOption();
    Properties properties = interpreterSetting.getProperties();
if (option.isExistingProcess) {
      properties.put(Constants.ZEPPELIN_INTERPRETER_HOST, option.getHost());
      properties.put(Constants.ZEPPELIN_INTERPRETER_PORT, option.getPort());
    }

//省略了同步等到相同key interpreterGroup终止的代码
    logger.info("Create interpreter instance {} for note {}", interpreterSetting.getName(), noteId);

for (String className : interpreterClassList) {
      Set<String> keys = Interpreter.registeredInterpreters.keySet();
for (String intName : keys) {
        RegisteredInterpreter info = Interpreter.registeredInterpreters.get(intName);
if (info.getClassName().equals(className)
            && info.getGroup().equals(groupName)) {
          Interpreter intp;

if (option.isRemote()) {//在单独启动的intepreter进程中创建Interpreter实例
            intp = createRemoteRepl(info.getPath(),
                key,
                info.getClassName(),
                properties,
                interpreterSetting.id());
          } else {//在zeppelin主进程中，动态加载并反射创建Interpreter实例
            intp = createRepl(info.getPath(),
                info.getClassName(),
                properties);
          }

synchronized (interpreterGroup) {
            List<Interpreter> interpreters = interpreterGroup.get(key);
if (interpreters == null) {
              interpreters = new LinkedList<Interpreter>();
              interpreterGroup.put(key, interpreters);//完成noteId:interpreters实例或者interpreterGroup：interpreters实例之间的映射
            }
            interpreters.add(intp);
          }
          logger.info("Interpreter " + intp.getClassName() + " " + intp.hashCode() + " created");
          intp.setInterpreterGroup(interpreterGroup);
break;
        }
      }
    }
  }

以下是createRemoteRepl方法实现，重点是用LazyOpenInterpreter Proxy了一个RemoteInterpreter实例：

  private Interpreter createRemoteRepl(String interpreterPath, String noteId, String className,
      Properties property, String interpreterSettingId) {
int connectTimeout = conf.getInt(ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT);
    String localRepoPath = conf.getInterpreterLocalRepoPath() + "/" + interpreterSettingId;
int maxPoolSize = conf.getInt(ConfVars.ZEPPELIN_INTERPRETER_MAX_POOL_SIZE);

    LazyOpenInterpreter intp = new LazyOpenInterpreter(new RemoteInterpreter(
        property, noteId, className, conf.getInterpreterRemoteRunnerPath(),
        interpreterPath, localRepoPath, connectTimeout,
        maxPoolSize, remoteInterpreterProcessListener));
return intp;
  }

InterpreterFactory执行解除note和interpreter实例之间的关系方法见removeInterpretersForNote：

  public void removeInterpretersForNote(InterpreterSetting interpreterSetting,
                                        String noteId) {
if (interpreterSetting.getOption().isPerNoteProcess()) {//perNoteProcess，直接关闭process
      interpreterSetting.closeAndRemoveInterpreterGroup(noteId);
    } else if (interpreterSetting.getOption().isPerNoteSession()) {//perNoteSession,由于process是共享的，note关闭，只关闭该note相关的interpreter实例，process不关闭
      InterpreterGroup interpreterGroup = interpreterSetting.getInterpreterGroup(noteId);

      interpreterGroup.close(noteId);
      interpreterGroup.destroy(noteId);
synchronized (interpreterGroup) {
        interpreterGroup.remove(noteId);
        interpreterGroup.notifyAll(); // notify createInterpreterForNote()
      }
      logger.info("Interpreter instance {} for note {} is removed",
          interpreterSetting.getName(),
          noteId);
    }
  }

参考文献

[1.] Apache Zeppelin, Interpreter mode explained, Lee Moon Soo