Flink SQL 1.11 on Zeppelin集成指南 - 简书


本站和网页 https://www.jianshu.com/p/5cdd2ab4f950 的作者无关,不对其内容负责。快照谨为网络故障时之索引,不代表被搜索网站的即时页面。

Flink SQL 1.11 on Zeppelin集成指南 - 简书登录注册写文章首页下载APP会员IT技术Flink SQL 1.11 on Zeppelin集成指南LittleMagic关注赞赏支持Flink SQL 1.11 on Zeppelin集成指南前言
大数据领域SQL化开发的风潮方兴未艾(所谓"Everybody knows SQL"),Flink自然也不能“免俗”。Flink SQL是Flink系统内部最高级别的API,也是流批一体思想的集大成者。用户可以通过简单明了的SQL语句像查表一样执行流任务或批任务,屏蔽了底层DataStream/DataSet API的复杂细节,降低了使用门槛。
但是,Flink SQL的默认开发方式是通过Java/Scala API编写,与纯SQL化、平台化的目标相去甚远。目前官方提供的Flink SQL Client仅能在配备Flink客户端的本地使用,局限性很大。而Ververica开源的Flink SQL Gateway组件是基于REST API的,仍然需要二次开发才能供给上层使用,并不是很方便。
鉴于有很多企业都无法配备专门的团队来解决Flink SQL平台化的问题,那么到底有没有一个开源的、开箱即用的、功能相对完善的组件呢?答案就是本文的主角——Apache Zeppelin。
Flink SQL on Zeppelin!
Zeppelin是基于Web的交互式数据分析笔记本,支持SQL、Scala、Python等语言。Zeppelin通过插件化的Interpreter(解释器)来解析用户提交的代码,并将其转化到对应的后端(计算框架、数据库等)执行,灵活性很高。其架构简图如下所示。
Flink Interpreter就是Zeppelin原生支持的众多Interpreters之一。只要配置好Flink Interpreter以及相关的执行环境,我们就可以将Zeppelin用作Flink SQL作业的开发平台了(当然,Scala和Python也是没问题的)。接下来本文就逐步介绍Flink on Zeppelin的集成方法。
配置Zeppelin
目前Zeppelin的最新版本是0.9.0-preview2,可以在官网下载包含所有Interpreters的zeppelin-0.9.0-preview2-bin-all.tgz,并解压到服务器的合适位置。
接下来进入conf目录。将环境配置文件zeppelin-env.sh.template更名为zeppelin-env.sh,并修改:
# JDK目录
export JAVA_HOME=/opt/jdk1.8.0_172
# 方便之后配置Interpreter on YARN模式。注意必须安装Hadoop,且hadoop必须配置在系统环境变量PATH中
export USE_HADOOP=true
# Hadoop配置文件目录
export HADOOP_CONF_DIR=/etc/hadoop/hadoop-conf
将服务配置文件zeppelin-site.xml.template更名为zeppelin-site.xml,并修改:
<!-- 服务地址。默认为127.0.0.1,改为0.0.0.0使得可以在外部访问 -->
<property>
<name>zeppelin.server.addr</name>
<value>0.0.0.0</value>
<description>Server binding address</description>
</property>
<!-- 服务端口。默认为8080,如果已占用,可以修改之 -->
<property>
<name>zeppelin.server.port</name>
<value>18080</value>
<description>Server port.</description>
</property>
最基础的配置就完成了。运行bin/zeppelin-daemon.sh start命令,返回Zeppelin start [ OK ]的提示之后,访问<服务器地址>:18080,出现下面的页面,就表示Zeppelin服务启动成功。
当然,为了一步到位适应生产环境,也可以适当修改zeppelin-site.xml中的以下参数:
<!-- 将Notebook repo更改为HDFS存储 -->
<property>
<name>zeppelin.notebook.storage</name>
<value>org.apache.zeppelin.notebook.repo.FileSystemNotebookRepo</value>
<description>Hadoop compatible file system notebook persistence layer implementation, such as local file system, hdfs, azure wasb, s3 and etc.</description>
</property>
<!-- Notebook在HDFS上的存储路径 -->
<property>
<name>zeppelin.notebook.dir</name>
<value>/zeppelin/notebook</value>
<description>path or URI for notebook persist</description>
</property>
<!-- 启用Zeppelin的恢复功能。当Zeppelin服务挂掉并重启之后,能连接到原来运行的Interpreter -->
<property>
<name>zeppelin.recovery.storage.class</name>
<value>org.apache.zeppelin.interpreter.recovery.FileSystemRecoveryStorage</value>
<description>ReoveryStorage implementation based on hadoop FileSystem</description>
</property>
<!-- Zeppelin恢复元数据在HDFS上的存储路径 -->
<property>
<name>zeppelin.recovery.dir</name>
<value>/zeppelin/recovery</value>
<description>Location where recovery metadata is stored</description>
</property>
<!-- 禁止使用匿名用户 -->
<property>
<name>zeppelin.anonymous.allowed</name>
<value>true</value>
<description>Anonymous user allowed by default</description>
</property>
Zeppelin集成了Shiro实现权限管理。禁止使用匿名用户之后,可以在conf目录下的shiro.ini中配置用户名、密码、角色等,不再赘述。注意每次修改配置都需要运行bin/zeppelin-daemon.sh restart重启Zeppelin服务。
配置Flink Interpreter on YARN
在使用Flink Interpreter之前,我们有必要对它进行配置,使Flink作业和Interpreter本身在YARN环境中运行。
点击首页用户名区域菜单中的Interpreter项(上一节图中已经示出),搜索flink,就可以看到参数列表。
Interpreter Binding
首先,将Interpreter Binding模式修改为Isolated per Note,如下图所示。
在这种模式下,每个Note在执行时会分别启动Interpreter进程,类似于Flink on YARN的Per-job模式,最符合生产环境的需要。
Flink on YARN参数
以下是需要修改的部分基础参数。注意这些参数也可以在Note中指定,每个作业自己的配置会覆盖掉这里的默认配置。
FLINK_HOME:Flink 1.11所在的目录;
HADOOP_CONF_DIR:Hadoop配置文件所在的目录;
flink.execution.mode:Flink作业的执行模式,指定为yarn以启用Flink on YARN;
flink.jm.memory:JobManager的内存量(MB);
flink.tm.memory:TaskManager的内存量(MB);
flink.tm.slot:TaskManager的Slot数;
flink.yarn.appName:YARN Application的默认名称;
flink.yarn.queue:提交作业的默认YARN队列。
Hive Integration参数
如果我们想访问Hive数据,以及用HiveCatalog管理Flink SQL的元数据,还需要配置与Hive的集成。
HIVE_CONF_DIR:Hive配置文件(hive-site.xml)所在的目录;
zeppelin.flink.enableHive:设为true以启用Hive Integration;
zeppelin.flink.hive.version:Hive版本号。
复制与Hive Integration相关的依赖到$FLINK_HOME/lib目录下,包括:
flink-connector-hive_2.11-1.11.0.jar
flink-hadoop-compatibility_2.11-1.11.0.jar
hive-exec-*.*.jar
如果Hive版本是1.x,还需要额外加入hive-metastore-1.*.jar、libfb303-0.9.2.jar和libthrift-0.9.2.jar
保证Hive元数据服务(Metastore)启动。注意不能是Embedded模式,即必须以外部数据库(MySQL、Postgres等)作为元数据存储。
Interpreter on YARN参数
在默认情况下,Interpreter进程是在部署Zeppelin服务的节点上启动的。随着提交的任务越来越多,就会出现单点问题。因此我们需要让Interpreter也在YARN上运行,如下图所示。
zeppelin.interpreter.yarn.resource.cores:Interpreter Container占用的vCore数量;
zeppelin.interpreter.yarn.resource.memory:Interpreter Container占用的内存量(MB);
zeppelin.interpreter.yarn.queue:Interpreter所处的YARN队列名称。
配置完成之后,Flink on Zeppelin集成完毕,可以测试一下了。
测试Flink SQL on Zeppelin
创建一个Note,Interpreter指定为flink。然后写入第一个Paragraph:
以%flink.conf标记的Paragraph用于指定这个Note中的作业配置,支持Flink的所有配置参数(参见Flink官网)。另外,flink.execution.packages参数支持以Maven GAV坐标的方式引入外部依赖项。
接下来创建第二个Paragraph,创建Kafka流表:
%flink.ssql表示利用StreamTableEnvironment执行流处理SQL,相对地,%flink.bsql表示利用BatchTableEnvironment执行批处理SQL。注意表参数中的properties.bootstrap.servers利用了Zeppelin Credentials来填写,方便不同作业之间复用。
执行上述SQL之后会输出信息:
同时在Hive中可以看到该表的元数据。
最后写第三个Paragraph,从流表中查询,并实时展现出来:
点击右上角的FLINK JOB标记,可以打开作业的Web UI。上述作业的JobGraph如下。
除SELECT查询外,通过Zeppelin也可以执行INSERT查询,实现更加丰富的功能。关于Flink SQL on Zeppelin的更多应用,笔者在今后的文章中会继续讲解,今天就到这里吧。
The End
民那晚安晚安。
评论12赞1414赞15赞赞赏更多好文

Copyright ©uecom 京ICP备18064371号-1 IPV6
2024-03-29 04:16:14
uenu.com.cn
10.0.12.16