Jupyter在美团民宿的应用实践

做算法的同学对于Kaggle应该都不陌生，除了举办算法挑战赛以外，它还提供了一个学习、练习数据分析和算法开发的平台。Kaggle提供了Kaggle Kernels，方便用户进行数据分析以及经验分享。在Kaggle Kernels中，你可以Fork别人分享的结果进行复现或者进一步分析，也可以新建一个Kernel进行数据分析和算法开发。Kaggle Kernels还提供了一个配置好的环境，以及比赛的数据集，帮你从配置本地环境中解放出来。Kaggle Kernels提供给你的是一个运行在浏览器中的Jupyter，你可以在上面进行交互式的执行代码、探索数据、训练模型等等。更多关于Kaggle Kernels的使用方法可以参考 Introduction to Kaggle Kernels ，这里不再多做阐述。

对于比赛类的任务，使用Kaggle Kernels非常方便，但我们平时的主要任务还是集中在分析、处理业务数据的层面，这些数据通常比较机密并且数量巨大，所以就不能在Kaggle Kernels上进行此类分析。因此，大型的互联网公司非常有必要开发并维护集团内部的一套「Kaggle Kernels」服务，从而有效地提升算法同学的日常开发效率。

本文我们将分享美团民宿团队是如何搭建自己的「Kaggle Kernels」—— 一个平台化的Jupyter，接入了大数据和分布式计算集群，用于业务数据分析和算法开发。希望能为有同样需求的读者带来一些启发。

现有系统与问题

算法同学在离线阶段主要包含三类任务：数据分析、数据生产、模型训练。为满足这些任务的要求，美团内部也开发了相应的系统：

• 魔数平台：用于执行SQL查询，下载结果集的系统。通常在数据分析阶段使用。
• 协同平台：用于使用SQL开发ETL的平台。通常用于数据生产。
• 托管平台：用于管理和运行Spark任务，用户提供任务的代码仓库，系统管理和运行任务。通常用于逻辑较复杂的ETL、基于Spark的离线模型训练/预测任务等。
• 调度平台：用于管理任务的依赖关系，周期性按依赖执行调度任务。

这些系统对于确定的任务完成的比较好。例如：当取数任务确定时，适合在魔数平台执行查询；当Spark任务开发就绪后，适合在托管平台托管该任务。但对于探索性、分析性的任务没有比较好的工具支持。探索性的任务有程序开发时的调试和对陌生数据的探查，分析性的任务有特征分析、Bad Case分析等等。

以数据探索为例，我们经常需要对数据进行统计与可视化，现有的做法通常是：魔数执行SQL -> 下载Excel -> 可视化。这种方式存在的问题是：

• 分析和取数工具割裂。
• 大数据分析可视化困难。

以Bad Case分析为例，现有的做法通常是：

这种方式存在的问题是：

• 分析与取数割裂，整个过程需要较多的手工操作。
• 分析过程不容易复现，对于多人协作式的验证以及进一步分析不利。
• 本地Python环境可能与分析对象的依赖有冲突，需要付出额外精力管理Python环境。

离线数据相关任务的模式通常是取数（小数据/大数据）–> Python处理（单机/分布式）–> 查看结果（表格/可视化）这样的循环。我们希望支持这一类任务的工具具有如下特质：

• 体验流畅：数据任务可以在统一的工具中完成，或者在可组合的工具链中完成。
• 体验一致：数据任务所用工具应该是一致的，不需要根据任务切换不同工具。
• 使用便捷：工具应是开箱即用，不需要繁琐的前置配置。
• 结果可复现：分析过程能够作为可执行代码保存下来，需要复现时执行即可，也应支持修改。

探索和分析类任务往往会带来可以沉淀的结果，如产生新的特征、模型、例行报告，希望可以建立起分析任务和调度任务的桥梁。

我们需要怎样的Jupyter

参考Kaggle Kernels的体验和开源Jupyter的功能，Notebook方式进行探索分析具有良好的体验。我们计划定制Jupyter，使其成为完成数据任务的统一工具。

这个定制的Jupyter应具备以下功能：

• 接入Spark：取数与分析均在Jupyter中完成，达到流畅、一致的体验。
• 接入调度系统：方便沉淀分析结果。
• 接入学城系统（内部WiKi）：方便分享和复现。
• 预配置环境：提供给用户开箱即用的环境。
• 用户隔离环境：避免用户间互相污染环境。

Jupyter项目架构

Project Jupyter由多个子项目组成，通过这些子项目可以自由组合出不同的应用。子项目的依赖关系如下图所示：

这个案例中，Jupyter应用是一个Web服务，我们可以从这个维度来看Jupyter架构：

Jupyter扩展方式

整个Jupyter项目的模块化和扩展性上都非常优秀。上图中的JupyterLab、Notebook Server、IPython、JupyterHub都是可扩展的。

JupyterLab扩展（labextension）

JupyterLab是Jupyter全新的前端项目，这个项目有非常明确的扩展规范以及丰富的扩展方式。通过开发JupyterLab扩展，可以为前端界面增加新功能，例如新的文件类型打开/编辑支持、Notebook工具栏增加新的按钮、菜单栏增加新的菜单项等等。JupyterLab上的前端模块具有非常清楚的定义和文档，每个模块都可以通过插件获取，进行方法调用，获取必要的信息以及执行必要的动作。我们在提供分享功能、调度功能时，均开发了JupyterLab扩展。JupyterLab扩展通常采用TypeScript开发，开发文档可参考： https://jupyterlab.readthedocs.io/en/stable/developer/extension_dev.html 。

JupyterLab核心组件依赖图

Notebook Server扩展（serverextension）

Notebook Server是用Python写的一个基于Tornado的Web服务。通过Notebook Server扩展，可以为这个Web服务增加新的Handler。增加新的Handler通常有两种用途：

1. 为JupyterLab扩展提供对应的后端接口，用于响应一些需要由服务端处理的事件。例如调度任务的注册需要通过JupyterLab扩展发起请求，由Notebook Server扩展执行。
2. 提供一个前端界面以及对应的后端处理服务。例如jupyter-rsession-proxy，用于在JupyterHub中使用RStudio。

Notebook Server扩展开发文档可参考： https://jupyter-notebook.readthedocs.io/en/stable/extending/handlers.html 。

Jupyter Kernels

Jupyter用于执行代码的模块叫Kernel，除了默认的ipykernel以外，还可以有其他的Kernel用于支持其他编程语言。例如支持Scala语言的almond、支持R语言的irkernel，更多详见 语言支持列表 。

IPython Magics

IPython Magics就是那些%、%%开头的命令。常见的Magics有 %matplotlib inline，设置Notebook中调用matplotlib的绘图函数时，直接展示图表在Notebook中。执行Magics时，事实上是调用了该Magics定义的一个函数。对于Line Magics（一个%），传入函数的是当前行的代码；对于Cell Magics（两个%），传入的是整个Cell的内容。定义一个新的IPython Magics仅需定义一个函数，这个函数的入参有两个，一个是当前会话实例，可以用来遍历当前会话的所有变量，可以为当前会话增加新的变量；另一个是用户输入，对于Line Magics是当前行，对于Cell Magcis是当前Cell。

IPython Magics在简化代码方面非常有效，我们开发了%%spark、%%sql用于创建Spark会话以及SQL查询。另外很多第三方的Magics可以用来提高我们的开发效率，例如在开发Word2Vec变种时，使用%%cython来进行Cython和Python混合编程，省去编译加载模块的工作。

IPython Magics开发文档可参考： https://ipython.readthedocs.io/en/stable/config/custommagics.html 。

IPython Widgets（ipywidgets）

IPython Widgets是一种基于Jupyter Notebook和IPython的可交互控件。与普通可视化不同的是，在控件上的交互会触发和Python的通信并执行相应的代码，Python上相应的动作也会触发界面实时变化。

IPython Widgets在提供工具类型的功能增强上非常有用，基于它，我们实现了一个线上排序服务的调试和复现工具，用于展示排序结果以及指定房源在排序过程中的各种特征以及中间变量的值。IPython Widgets的开发可以通过组合现有的Widgets实现，也可以完全自定义一个，IPython Widgets开发文档可参考： https://ipywidgets.readthedocs.io/en/stable/examples/Widget%20Custom.html 。

ipyleaflet

扩展JupyterHub

Authenticators

JupyterHub是一个多用户系统，登录模块可替换，通过实现新的Authenticator类并在配置文件中指定即可。通过这个扩展点，我们实现了使用内部SSO系统登录JupyterHub。Authenticator开发文档可参考： https://jupyterhub.readthedocs.io/en/stable/reference/authenticators.html 。

Spawners

当用户登录时，JupyterHub需要为用户启动一个用户专用Notebook Server。启动这个Notebook Server有多种方式：本机新的Notebook Server进程、本机启动Docker实例、K8s系统中启动新的Pod、YARN中启动新的实例等等。每一种启动方式都对应一个Spawner，官方提供了多种Spawner的实现，这些实现本身是可配置的。如果不符合需求，也可以自己开发全新的Spawner。由于我们需要实现Spark接入，对K8s的Pod有新的要求，所以基于KubeSpawner定制了一个Spawner来解决Spark连接集群的网络问题。Spawner开发文档可参考： https://jupyterhub.readthedocs.io/en/stable/reference/spawners.html 。

我们的定制

回顾我们的需求，这个定制的Jupyter应具备以下功能：

• 接入Spark：可以通过配置容器环境以及Spawner完成。
• 接入调度系统：需要开发JupyterLab扩展以及Notebook Server扩展。
• 接入学城系统：需要开发JupyterLab扩展以及Notebook Server扩展。
• 预配置环境：镜像配置。
• 用户隔离环境：通过定制Authenticators + K8s Spawner实现容器级别环境隔离。

我们的方案是基于JupyterHub on K8s。下图是平台化Jupyter的架构图，从上到下可以看到三条主线：1. 分享复现、2. 探索执行、3. 调度执行。

几个关键组件介绍：

• JupyterLab：交互式执行的前端，开源项目。
• Jupyter Server：交互式执行的后端，开源项目。
• Commuter：浏览Notebook的工具，开源项目。
• K8s：容器编排系统，开源项目。
• Cantor：美团调度系统，同类开源项目有AirFlow。
• 托管平台：美团离线任务托管平台，给定代码仓库和任务参数，为我们执行Spark-Submit的平台。
• 学城：美团文档系统。
• MSS：美团对象存储。
• NB-Runner：Notebook Runner，在nbconvert的基础上增加了参数化和Spark支持。

在定制Jupyter中，最为关键的两个是接入Spark以及接入调度系统，下文中将详细介绍这两部分的原理。

JupyterHub on K8s包括几个重要组成部分：Proxy、Hub、Kubernetes、用户容器（Jupyter Server Pod）、单点登录系统（SSO）。一个用户在登录后新建容器实例的过程中，这几个模块的交互如下图所示：

可以看到，新建容器实例后，用户的交互都是经过Proxy后与Jupyter Server Pod进行通信。因此，扩展功能的工作主要是定制Jupyter Server Pod对应的容器镜像。

让Jupyter支持Spark

Jupyter平台化后，我们得到一个接近Kaggle Kernel的环境，但是还不能够使用大数据集群。接下来，就是让Jupyter支持Spark，Jupyter支持Spark的方案有Toree，出于灵活性考虑，我们没有使用。我们希望让普通的Python Kernel能支持PySpark。

为了能让Jupyter支持Spark，我们需要了解两方面原理：Jupyter代码执行原理和PySpark原理。

Jupyter代码执行原理

所用到的Jupyter分三部分：前端JupyterLab、服务端Jupyter Server、语言Kernel IPython。这三个模块的通信如下图所示：

Jupyter执行代码时序图

这里，需要在IPython的exec阶段支持PySpark。

PySpark原理

启动PySpark有两种方式：

• 方案一：PySpark命令启动，内部执行了spark-submit命令。
• 方案二：任意Python shell（Python、IPython）中执行Spark会话创建语句。

这两种启动方式有什么区别呢？

看一下PySpark架构图：

PySpark架构图，来自 SlideShare

与Spark的区别是，多了一个Python进程，通过Py4J与Driver JVM进行通信。

PySpark方案启动流程

PySpark启动时序图

IPython方案启动流程

实际的IPython中启动Spark时序图

Toree采用的是类似方案一的方式，脚本中调用spark-submit执行特殊版本的Shell，内置了Spark会话。我们不希望这么做，是因为如果这样做的话就会：

• 多了一个PySpark专供的Kernel，我们希望Kernel应该是统一的IPython。
• PySpark启动参数是固定的，配置在kernel.json里。希望PySpark任务是可以按需启动，可以灵活配置所需的参数，如Queue、Memory、Cores。

因此我们采用方案二，只需要一些环境配置，就能顺利启动PySpark。另外为了简化Spark启动工作，我们还开发了IPython的Magics，%spark和%sql。

环境配置

为了让IPython中能够顺利启动起Spark会话，需要正确配置如下环境变量：

• JAVA_HOME：Java安装路径，如/usr/local/jdk1.8.0_201。
• HADOOP_HOME：Hadoop安装路径，如/opt/hadoop。
• SPARK_HOME：Spark安装路径，如/opt/spark-2.2。
• PYTHONPATH：额外的Python库路径，如$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.4-src.zip。
• PYSPARK_PYTHON：集群中使用的Python路径，如./ARCHIVE/notebook/bin/python。集群中使用Python通常需要虚拟环境，通过spark.yarn.dist.archives带上去。
• PYSPARK_DRIVER_PYTHON：Spark Driver所用的Python路径，如果你用Conda管理Python环境，那这个变量应为类似/opt/conda/envs/notebook/bin/python的路径。

为了方便，建议设置各bin路径到PATH环境变量中：$SPARK_HOME/sbin:$SPARK_HOME/bin:$HADOOP_HOME/sbin:$HADOOP_HOME/bin:$JAVA_HOME/bin:$PATH。

完成这些之后，可以在IPython中执行创建Spark会话代码验证：

import pyspark
spark = pyspark.sql.SparkSession.builder.appName("MyApp").getOrCreate()

在Spark任务中执行Notebook

执行Notebook的方案目前有nbconvert，Python API方式执行样例如下所示，暂时称这段代码为NB-Runner.py：

# Import：首先我们import nbconvert和ExecutePreprocessor类：
import nbformat
from nbconvert.preprocessors import ExecutePreprocessor

# 加载：假设notebook_filename是notebook的路径，我们可以这样加载：
with open(notebook_filename) as f:
    nb = nbformat.read(f, as_version=4)

# 配置：接下来，我们配置notebook执行模式：
ep = ExecutePreprocessor(timeout=600, kernel_name='python')

# 执行(preprocess)：真正执行notebook的地方是调用函数preprocess：
ep.preprocess(nb, {'metadata': {'path': 'notebooks/'}})

#保存：最后，我们保存notebook执行结果：
with open('executed_notebook.ipynb', 'w', encoding='utf-8') as f:
    nbformat.write(nb, f)

现在有两个问题需要确认：

• 当Notebook中存在Spark相关代码时，Python NB-Runner.py能否正常执行？
• 当Notebook中存在Spark相关代码时，Spark-Submit NB-Runner.py能否正常执行？

之所以会出现问题2，是因为我们的调度系统只能调度Spark任务，所以必须使用Spark-Submit的方式来启动NB-Runner.py。

为了回答这两个问题，需要了解nbconvert是如何执行Notebook的。

nbconvert执行时序图

问题1从原理上看，是可以正常执行的。实际测试也是如此。对于问题2，答案似乎并不明显。结合“PySpark启动时序图”、“实际的IPython中启动Spark时序图”与“nbconvert执行时序图”：

Spark-Submit NB-Runner.py的方式存在问题的点可能在于，IPython中执行Spark.builder.getOrCreate时，Driver JVM已经启动并且Py4J Gateway Server已经实例化完成。如何让Spark.builder.getOrCreate执行时跳过上图“实际的IPython中启动Spark时序图”的Popen(spark-submit)以及后续的启动Py4J Gateway Server部分，直接与Py4J Gateway Server建立连接？

在PySpark代码中，看到如下这段代码：

def launch_gateway(conf=None):
    """
    launch jvm gateway
    :param conf: spark configuration passed to spark-submit
    :return:
    """
    if "PYSPARK_GATEWAY_PORT" in os.environ:
        gateway_port = int(os.environ["PYSPARK_GATEWAY_PORT"])
    else:
        SPARK_HOME = _find_spark_home()
        # Launch the Py4j gateway using Spark's run command so that we pick up the
        # proper classpath and settings from spark-env.sh
        on_windows = platform.system() == "Windows"
        script = "./bin/spark-submit.cmd" if on_windows else "./bin/spark-submit"
...

如果我们能在IPython进程中设置环境变量PYSPARK_GATEWAY_PORT为真实的Py4J Gateway Server监听的端口，就会跳过Spark-Submit以及启动Py4J Gateway Server部分。那么PYSPARK_GATEWAY_PORT从哪来呢？我们发现在Python进程中存在这个环境变量，只需要通过ExecutorPreprocessor将它传递给IPython进程即可。

数据分析与可视化

数据探查和数据分析在这里都是同样的流程。用户要分析的数据通常存储在MySQL和Hive中。为了方便用户在Notebook中交互式的执行SQL，我们开发了IPython Magics %%sql用来执行SQL。

SQL Magics的用法如下：

%%sql <var> [--preview] [--cache] [--quiet]
SELECT field1, field2
  FROM table1
 WHERE field3 == field4

SQL查询的结果暂存在指定的变量名 中，对于MySQL数据源 的类型是Pandas DataFrame，对于Hive数据源 的类型是Spark DataFrame。 可用于需要对结果集进行操作的场合，如多维分析、数据可视化。目前，我们支持几乎所有的Python数据可视化库。

下图是一个数据分析和可视化的例子：

数据分析与可视化

Notebook分享

Notebook不仅支持交互式的执行代码，对于文档编辑也有不错的支持。数据分析过程中的数据、表格、图表加上文字描述就是一个很好的报告。Jupyter服务还支持用户一键将Notebook分享到美团内部的学城中。

一键分享：

一键分享

上述数据分析分享到内部学城的效果如下图所示：

Notebook分享效果

模型训练

基于大数据的模型训练通常使用PySpark来完成。除了Spark内置的Spark ML可以使用以外，Jupyter服务上还支持使用第三方X-on-Spark的算法，如XGBoost-on-Spark、LightGBM-on-Spark。我们开发了IPython Magics %%spark来简化这个过程。

Spark Magics的用法如下：

%%spark
[--conf <property-name>=<property-value>]
[--conf <property-name>=<property-value>]
...

执行%%spark后，会启动Spark会话，启动后Notebook会话中会新建两个变量spark和sc，分别对应当前Spark会话的SparkSession和SparkContext。

下图是一个使用LightGBM-on-Yarn训练模型的例子，基于Azure/mmlspark官方Notebook例子，仅需添加启动Spark语句以及修改数据集路径。

LightGBM on Spark Demo

排序策略调试

通过开发ipywidgets实现了一个线上排序策略的调试工具，可以用于查看排序结果以及排序原因（通过查看变量值）。

通过平台化Jupyter的定制与部署，我们实现了数据分析、数据生产、模型训练的统一开发环境。在此基础上，还集成了内部公共服务和业务服务，从而实现了从数据分析到策略上线到结果分析的全链路支持。

我们对这个项目未来的定位是数据科学的云端集成开发环境，而Jupyter项目所具有的极强扩展性，也能够支持我们朝着这个方向不断进行演进。

文龙，美团民宿研发团队工程师。 颖艺，美团民宿研发团队工程师。

美团民宿研发团队诚招数据系统研发工程师，Base厦门，欢迎有兴趣的同学投递简历到 [email protected] 。