Flink

更多Flink系列文章请点击Flink系列文章

更多大数据文章请点击大数据好文推荐

最近在调研流平台，发现各大公司流平台的Web界面都是自己一个团队开发，相当完备。苦于人力、时间有限，想找现成的能提交FlinkSql的Web代码，没找到合适的开源的。但是想起了之前看过的Zeppelin，现在已经支持Flink 1.0且支持DataStream、Table & SQL 等，遂赶紧尝试。如果能走通，后续计划在原数据这块儿看看有什么好办法管理起来。

Zeppelin基于界面化的笔记本，可实现数据驱动，使用SQL、Scala（没看错，能指直接写Scala代码！）等进行交互式数据分析。

可视化的笔记本是指：

Zeppelin Server是Zeppelin最主要的服务之一，负责管理Interpreter。可以启动多个Zeppelin Server组成服务状态一致的集群（使用了Raft协议，具体是 Raft算法库Atomix），共享Notebook、元数据，实现Interpreter负载均衡。

普通ZeppelinServer发送了本节点的CPU和内存资源平均使用情况，用以在创建Interpreter时进行分配。

Raft Leader ZeppelinServer会监控所有类型节点上报的心跳数据，发现超时就认为不可用并剔除。

可通过Nginx等技术进行前端代理。多个用户访问域名时，Nginx根据分布式策略来将用户分配到不同的可用的Zeppelin Server，如上图User1/2。

每个Interpreter进程都是一个JVM进程，通过thrift和Zeppelin Server交互。单节点模式时在ZeppelinServer本地创建，集群模式时先从集群元数据中查找是否已经存在所需Interpreter如果有就直接通过元数据中的该Interpreter进程的Thrift IP和端口来将note和该进程建立绑定关系；如果不存在就通过元数据找出最空闲的ZeppelinServer节点，并通过Thrift来告知远程ZeppelinServer创建Interpreter进程。

用户使用的notebook运行在Interpreter上，底层就能跑各种引擎、语言。

Interpreter 进程启动后，将会在 Zeppelin Cluster MetaData 中提交自身的元数据信息，关闭时清除。如果没有正常退出导致元数据没有先清理，ZeppelinServer会主动周期性检查该元数据对应的Interpreter的心跳时间戳来确定对应的进程是否存活。

Zeppelin Server和Interpreter进程的通信手段是Netty。

InterpreterGroup 可包含多个Interpreter，比如SparkInterpreter Group 包含了 SparkSqlInterpreter、SparkInterpreter、PySparkInterpreter 等Intepreter。

当用户使用 SparkInterpreter Group 创建一个 notebook 时，ZeppelinServer 会创建一个独立的 JVM 进程，进程中的 SparkSqlInterpreter 、 SparkInterpreter 、PySparkInterpreter可以共用一个 SparkContext，因为他们属于同一个InterpreterGroup。

InterpreterGroup和Interpreter关系具体还要取决于Interpreter Binding Mode。

集群中元数据信息，KV键值对格式，包括ZeppelinServer和Interpreter进程元数据，通过State Machine 维护服务状态一致性。

ZeppelinServer和Interpreter都会周期性发送心跳来更新Cluster Metadata中自己的信息，而担任Leader的那个ZeppelinServer会定时检查Cluster Metadata中的时间戳信息，如果有超时的就会清理超时的服务和进程。

用户工作的平台，包含若干Paragraph。

默认为local即存在本地，集群模式时应该选择所有集群中的Zeppelin节点都能访问的位置，比如HDFS。

集群模式下，会将修改自动同步到所有ZeppelinServer节点。

一个Notebook包含若干Paragraph，可以共享数据。

比如一个FlinkSQL程序，可以在定义3个Paragraph:

Zeppelin interpreter是个重要的组件，可将任何语言和数据处理后端以插件化的方式接入Zeppelin，目前支持的技术栈如下： 如果没有你需要的，那还可以自定义一个：

通过Interpreters，我们可以很方便的使用各种语言和数据处理后端，比如可以直接用%flink来直接在Zeepelin中写scala代码。

用来控制Notebook和用户的隔离模式：

Zeppelin 0.8.0以后提供了LifecycleManager接口来控制interpreter生命周期，0.9.0有两个实现：

点击这里

也是跟创建一样点击进入Interpreter界面，搜索后可以修改创建的interpreter。

见interpreter

Interpreter有很多配置，可直接在Zeppelin上Web界面上设置，有两类属性：

关于Context Parameters

上面说的都是interpreter下的通用配置，所有使用该interpreter的Notebook都是用该配置。但有些时候我们想每个notebook单独使用某些配置，虽然可创建单独interpreter但很不方便，所以可使用Inline Generic Configuration。

即在notebook最开始的paragraph里面写:

可通过interpreter 配置zeppelin.interpreter.launcher yarn来讲interpreter运行在yarn上，然后会和Zeppelin服务端通过Thrift交互。

具体源码在YarnInterpreterLauncher和YarnRemoteInterpreterProcess。

可配合flink on yarn使用，则Interpreter在yarn上am 中拉起来后，会去拉起一个flink cluster。相关源码在FlinkScalaInterpreter、FlinkInterpreter等。

flink配置文件上传和读取流程如下：

Download Apache Zeppelin

有三种方式：下载完整已编译二进制包、通过网络安装指定interpreter的包或者源码自己编译安装。

图方便就用第一种吧（不过很大，有1.5G）。

可参考

必须是JDK 1.8(171)以上

下载完后直接解压zeppelin-0.9.0-preview1-bin-all.tgz。

可参考

可修改两个配置文件，都配置了同key属性时以环境变量文件为准：

我改了几个属性：

zeppelin.server.addr 改为ip，默认127.0.0.1，不然其他机器访问不了

zeppelin.server.port zeepelin启动后的web端口

zeppelin.interpreter.lifecyclemanager.class 改为org.apache.zeppelin.interpreter.lifecycle.TimeoutLifecycleManager，默认是NullLifecycleManager，不会管interpreter是否空闲。而TimeoutLifecycleManager会在interpreter保持空闲状态超过zeppelin.interpreter.lifecyclemanager.timeout.threshold毫秒时，干掉interpreter。

zeppelin.interpreter.lifecyclemanager.timeout.threshold 改为 3600000

zeppelin.interpreter.lifecyclemanager.timeout.checkinterval 改为60000，检测interpreter是否超时的间隔时间

zeppelin.recovery.storage.class 指定zeppelin恢复模式(详见)。

改为org.apache.zeppelin.interpreter.recovery.FileSystemRecoveryStorage。设定后，关掉Zeppelin主进程不会关掉interpreter进程，重启zeppelin会去重连这些interpreter进程。

这个时候如果还想干掉所有interpreter进程，请使用bin/stop-interpreter.sh。

默认NullRecoveryStorage，意味着关掉Zeppelin就关掉了所有运行中的interpreter 进程。

zeppelin.recovery.dir 在集群模式下，还应该把此项设为hdfs上路径，如/tmp/zeppelin/recovery。注意不要加如hdfs://namespace，否则路径不对！ 将Notebook保存在共享存储中

zeppelin.notebook.storage 设为org.apache.zeppelin.notebook.repo.FileSystemNotebookRepo

zeppelin.notebook.dir 设为hdfs上的目录，如/tmp/zeppelin/notebook。这里注意不能设为hdfs://namespace/xxx，否则路径会有问题！

可参考

重要配置如下：

flink.interpreter.close.shutdown_cluster 改为false。

默认true，即在interpreter关闭时会shutdown应用程序。

注意，在interpreter on yarn+flink on yarn模式下可以将此设置改为true，效果就是只要你需要重启interpreter来改配置就会停止对应的flink app，达成同步。

zeppelin.interpreter.close.cancel_job 改为false

默认true，即在interpreter关闭时会cancel我们的flink job。

注意，在interpreter on yarn+flink on yarn模式下可以将此设置改为true，效果就是只要你需要重启interpreter来改配置就会停止对应的flink app，达成同步。

如果flink任务依赖一些包，可以有三种方式加入依赖：

flink.execution.jars 指定flink job所依赖的普通jar包，所有的jar包都被会load到flink interpreter的classpath，还会被发送到Task Manager。

flink.udf.jars 和flink.execution.jars不同的地方在于Zeppelin会自动检测该选项指定的jar包中所包含的UDF class，会把检测到的UDF注册到TableEnvironment中（UDF的名字就是这个class name），以便用户使用。

注意：

flink.execution.packages 类似flink.execution.jars，但不同的是Zeppelin会下载该选项指定的package以及该package的依赖放到flink interpreter的classpath。比如你想使用kafka connector配置如下

但我实测在测试环境无问题，生产环境因为安全策略不能自动下载，只能去repo1.maven手动下载打好的jar（flink-connector-kafka_2.11-1.10.0.jar、flink-connector-kafka-base_2.11-1.10.0.jar、flink-sql-connector-kafka_2.11-1.10.0.jar、flink-json-1.10.0.jar，如果找不到包还可以放入kafka-clients-2.2.0.jar），并upload到使用的flink/lib下，否则会报错如下：

需要添加一些必要的jar包放在FLINK_HOME//lib下：

如果要采用Hive来存元数据或访问hive还需要:

最新master分支中，还可通过interpreter 配置zeppelin.interpreter.launcher yarn来讲interpreter运行在yarn上，然后会和Zeppelin服务端通过Thrift交互，并向yarn申请资源来启动flink cluster。详见Flink on Zeppelin  (7). Yarn Interpeter 模式。

此模式其他要求：

interpreter local + flink on yarn: interpreter on yarn + flink on yarn:

相关interpreter配置如下：

进入安装好的zeppelin目录后，执行

随后就可以访问Zeepelin Web界面了： 可以看到，已经有了一些现成的Notebook示例。

重要说明：

必须将Notebook保存在共享存储中

zeppelin.recovery.dir 在集群模式下，还应该把此项设为hdfs上路径，如/tmp/zeppelin/recovery。注意不要加如hdfs://namespace，否则路径不对！

zeppelin.cluster.addr 配置集群中的所有ZeppelinServerIP及Raft端口，Raft据此进行Leader选举、元数据维护等。多个地址用逗号分隔。

这里需要注意的是，Raft端口号配置不要和zeppelin.server.port配置相同，否则会造成冲突！

注意事项：

如果在Zeppelin集群模式下使用FlinkSql on Yarn，需要做如以下配置，否则报错：

该错误我已经提交了issue给社区，详情可见：ZEPPELIN-4809

主要是可以让Flink使用Hive Catalog存储Flink SQL 元数据（可参考HiveCatalog，注意这种表只能由Flink读写使用，不要用Hive去读写。可以在Hive命令行中使用DESCRIBE FORMATTED命令查看表的元数据，如果是is_generic=true代表是Flink专用表），也可以直接使用Flink读写Hive表数据。

需要将以下包放入$FLINK_HOME/lib:

然后设置flink interpreter:

随后，使用flinksql注册的表会自动保存到hive default库里。

可以使用flink 配置，实现0代码配置checkpoint。

具体请参考flink-checkpoint配置

zeppelin可采用LDAP做身份认证+shiro做权限控制。

相关内容可参考:

比如我们有一些ddl中定义了数据库连接信息，这些信息十分敏感不想暴露给其他人，这个时候我们可以用Credentials。

Credentials配置 先在interpreter配置injectCredentials true，也可在notebook界面做配置，比如执行时使用%flink(injectCredential=true)

Credentials打开

Credentials定义 这里的Entity就相当于是你的Credentials的Key，在访问时使用。

需要注意的是，每个人创建的Credentials对其他人都不可见，别人也无法使用。

Credentials使用 再次强调，每个人只能使用自己的Credentials。格式为user.EntityName和password.EntityName。（因0.9-preview尚不是稳定版本，可能存在一定变化，还可尝试EntityName.user）。

本用户使用Credentials效果： 用户使用其他人的Credentials效果，打出的是原始字符串而不是Credentials：

必须使用版本号3.5-3.7的python

pip install apache-flink 如果安装很慢或者超时，可以加参数pip --default-timeout=100000 install -i apache-flink。

如果提示pip需要升级，可执行pip --default-timeout=100000 install -i --upgrade pip

将flink-python_2.11-1.10.0.jar从$FLINK_HOME/opt移动到$FLINK_HOME/lib:

配置好flink_interpreter的zeppelin.pyflink.python为python路径

已经支持一些基本图标，任意后端输出都可以以图表方式展现！（不用后端人员再去学什么echarts了）

直接可以拖拽方式生成。

还想了解更多或二次开发就参考：

Dynamic Forms 是Zeppelin的一个高级功能，允许用户在代码中插入UI控件来允许用户定制化你的代码。Jdbc Interpreter支持这一功能，用户可以定制SQL，下面是一个下拉框的例子。

可以直接将你的Zeppelin笔记本url分享给其他写作者，则大家都可以看到实时更新。

如上图，如果ZeppelinServer1 突然挂掉不可用，如果正确配置了相关配置zeppelin.recovery.storage.class，则不会影响其上运行的interpreter进程，如果此时这些进程可访问，则其他ZeppelinServer节点可以通过Cluster Metadata读取到这些interpreter进程元数据信息，让用户继续使用这些interpreter进程。

当然Nginx也会发现ZeppelinServer1出现异常，将它视为离线状态。

那么现在本来是用ZeppelinServer1的User1再次启动NoteBook时，Nginx会将请求发送到其他ZeppelinServer节点来使用之前的interpreter进程。

当ZeppelinServer1所在节点整个挂掉时，其他ZeppelinServer会删除无效元数据，并重建interpreter进程。

Zeppelin前后端分离架构，可参考：

问题：

Apache Zeppelin公众号

钉钉讨论群

Flink-Zeppelin On FlinkSql