1. 为什么要在 Zeppelin 里跑 Flink SQL1.1 一个真实的需求场景先说一个我踩过的坑。早些年做实时数仓团队里数据开发同学写 Flink SQL 的方式基本是两种要么在 IDE 里写 Java/Scala 代码把 SQL 硬编码进去打包提交要么用 Flink 自带的 SQL Client 命令行一条一条敲。前者改一个字段就要重新编译打包后者在集群上调试时历史命令翻起来费劲结果也没法保存和分享。后来接触到 Zeppelin才发现它天生就是干这个的。Zeppelin 是一个基于 Web 的交互式笔记本工具最擅长的事情就是“写一段、跑一段、看结果”。它原生支持多种解释器Interpreter其中就包括 Flink。把 Flink SQL 放到 Zeppelin 里跑等于给 Flink 配了一个带界面、能保存、能可视化、能多人协作的 SQL 编辑器。这个组合解决的核心问题很明确让 Flink SQL 的开发从“命令行 打包”变成“浏览器里即写即跑”。适合谁用数据开发、实时计算工程师、做实时数仓的分析师以及任何需要频繁调试 Flink SQL 但又不想每次都走完整打包流程的人。哪怕你之前只写过 Hive SQL只要理解 Flink SQL 的基本语法这套环境也能很快上手。1.2 Zeppelin 与 Flink 的版本匹配是第一个坎在动手之前必须先说清楚一个最容易翻车的地方Zeppelin 版本和 Flink 版本的匹配。Zeppelin 的 Flink Interpreter 是跟着 Flink 版本走的不同大版本之间 API 差异很大硬凑会直接报ClassNotFoundException或者NoSuchMethodError。我整理了一张常见组合的对照表这是实测下来比较稳的搭配Zeppelin 版本推荐 Flink 版本说明0.10.x1.13.x / 1.14.x老项目常用稳定但功能偏旧0.11.x1.14.x / 1.15.x过渡版本社区资料较多0.12.x1.16.x / 1.17.x目前比较推荐的组合提示不要盲目追最新版。Zeppelin 的 Flink Interpreter 更新往往滞后于 Flink 主版本选一个社区验证过的组合比用最新版省心得多。另外要区分两种运行模式Local 模式Zeppelin 自带 Flink开箱即用适合学习和调试和Remote 模式连接一个独立部署的 Flink 集群适合生产。本文两种都会讲到但重点放在 Remote 模式因为实际工作中几乎都用它。2. 安装部署从零把环境搭起来2.1 前置环境准备在装 Zeppelin 之前机器上得先有几样东西。我列一下清单缺一样后面都可能卡住JDK 8 或 JDK 11Flink 1.15 之后对 JDK 11 支持更好但 JDK 8 兼容性最广。建议统一用 JDK 8避免各种奇怪的反射报错。一个可用的 Flink 集群Standalone 或基于资源调度框架的都可以。如果只是本地测试用 Flink 自带的start-cluster.sh起一个 Standalone 集群就够了。Hadoop 环境可选如果你的 Flink SQL 要读写 HDFS 或者用 Hive Catalog那 Hadoop 相关配置得先配好否则连接器加载会失败。环境变量这块JAVA_HOME必须配好HADOOP_CONF_DIR如果用到 HDFS 也要指向正确的配置目录。这些是基础但每年都能看到有人因为JAVA_HOME没配导致 Zeppelin 起不来。2.2 下载与解压 ZeppelinZeppelin 的安装包直接去官网下载对应版本即可。下载下来是个 tar 包解压到你想放的目录tar -zxvf zeppelin-0.12.0-bin-all.tgz -C /opt/ cd /opt/zeppelin-0.12.0-bin-all注意选bin-all这个包它包含了所有解释器包括 Flink。如果下的是bin-netinst那 Flink Interpreter 还得单独装徒增麻烦。解压完之后目录结构大致是这样bin放启动脚本conf放配置interpreter放各解释器的配置notebook放笔记本数据。先别急着启动配置改完再启动。2.3 配置 Flink Interpreter这是整个安装过程的核心。Zeppelin 的 Flink Interpreter 配置在conf/zeppelin-site.xml和解释器界面里都能改但最直接的方式是改conf/interpreter.json或者启动后在 Web UI 里配。我习惯先改配置文件因为 Web UI 改完有时候重启会丢。关键配置项有这么几个FLINK_HOME指向你的 Flink 安装目录。Remote 模式下这个必须对因为 Zeppelin 要从这里加载 Flink 的客户端 jar。zeppelin.flink.mode取值local或remote。生产环境填remote。zeppelin.flink.jobmanager.host和zeppelin.flink.jobmanager.portRemote 模式下指向 Flink JobManager 的地址和 RPC 端口默认 6123。zeppelin.flink.scala.version根据你的 Flink 版本选 2.11 或 2.12选错了会报 Scala 版本冲突。配置改完后把 Flink 的 jar 包软链或者拷贝到 Zeppelin 的 interpreter 目录下。具体来说需要把$FLINK_HOME/lib下的核心 jar 和$FLINK_HOME/opt下你需要的连接器 jar 放到 Zeppelin 能加载到的路径。这一步很多人会漏结果就是 SQL 能解析但一执行就报找不到连接器。2.4 启动与验证配置就绪后启动 Zeppelinbin/zeppelin-daemon.sh start启动日志在logs/zeppelin-*.log如果起不来第一件事就是看这个日志。常见问题无非是端口占用默认 8080、JDK 版本不对、或者内存不够。启动成功后浏览器打开http://你的机器IP:8080能看到 Zeppelin 的首页。点右上角Create new note在 Interpreter 下拉框里选flink如果能看到这个选项说明 Flink Interpreter 已经加载成功了。注意第一次创建 Flink 笔记本时Zeppelin 会去初始化 Flink 环境这个过程可能比较慢耐心等它跑完。如果一直卡住多半是FLINK_HOME配错了或者 jar 没放对位置。3. Flink SQL 在 Zeppelin 里的实操写法3.1 基本执行流程与段落概念Zeppelin 的核心概念是段落Paragraph。一个笔记本里可以有多个段落每个段落独立执行执行结果直接显示在段落下方。这个设计对 Flink SQL 特别友好因为你可以把“建表”“插入数据”“查询验证”拆成不同段落改哪段跑哪段不用从头再来。在 Flink 笔记本里默认的段落开头是%flink表示用 Flink Interpreter 执行。但 Flink SQL 有几种不同的执行入口需要显式声明%flink.ssql执行流式 SQLStreaming SQL最常用。%flink.batch执行批式 SQL。%flink.bsql批式 SQL 的另一种写法。%flink.pyflink用 Python 写 Flink 任务。%flink执行 Scala/Java 代码用来做 SQL 之外的操作。我平时用得最多的是%flink.ssql因为实时场景基本都是流式的。写一个最简单的例子%flink.ssql CREATE TABLE source_table ( id BIGINT, name STRING, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL 5 SECOND ) WITH ( connector datagen, rows-per-second 10 ); SELECT * FROM source_table;这段代码用datagen连接器造数据跑起来后 Zeppelin 会以表格形式实时刷新结果非常直观。这就是 Zeppelin 相比 SQL Client 的最大优势——结果可视化。3.2 建表与连接器配置的细节Flink SQL 的建表语句里WITH子句是连接器配置也是最容易出错的地方。不同连接器的参数差异很大我挑几个常用的说一下。Kafka 连接器是实时场景的标配。关键参数包括connector、topic、properties.bootstrap.servers、format。如果 Kafka 开了认证还要配properties.security.protocol和相关的用户名密码参数。这里有个坑properties.前缀不能漏漏了参数不会生效但也不报错数据就是读不进来排查起来很费劲。JDBC 连接器用于读写关系型数据库。url、table-name、username、password是必填的。如果做维表关联还要注意lookup.cache.max-rows和lookup.cache.ttl这两个参数它们决定了维表缓存的策略配不好会导致数据延迟或者内存暴涨。Hive 连接器需要先配置 Hive Catalog。在 Zeppelin 里可以通过%flink段落执行 Scala 代码来注册 Catalog%flink val catalog new org.apache.flink.table.catalog.hive.HiveCatalog(hive, default, /path/to/hive/conf) tenv.registerCatalog(hive, catalog) tenv.useCatalog(hive)注册完之后在%flink.ssql段落里就能直接查 Hive 表了。这个能力在做流批一体的时候特别有用。3.3 流式查询与结果展示Zeppelin 对 Flink 流式查询的结果展示做了专门优化。默认情况下流式查询的结果会以“追加模式”不断刷新。但有些查询比如聚合需要“更新模式”这时候要在 SQL 里显式指定%flink.ssql(typeupdate) SELECT name, COUNT(*) AS cnt FROM source_table GROUP BY name;typeupdate这个参数告诉 Zeppelin 用更新模式展示结果否则聚合查询会报错说需要更新模式。这个细节很多新手不知道卡在这里半天。另外Zeppelin 还支持把流式结果画成图表。在结果区域上方有个图表按钮可以选折线图、柱状图等。对于监控类的实时指标这个功能直接省掉了一个可视化面板的开发。实操心得流式查询在 Zeppelin 里跑起来后会一直占用资源。调试完记得手动停止段落否则它会一直跑着消耗集群资源。我见过有人忘了停第二天发现集群被一个测试任务占满了。4. 常见问题与排查技巧实录4.1 启动与连接类问题问题一Zeppelin 启动后看不到 flink 解释器选项。这个基本是 Flink Interpreter 没加载成功。排查顺序先看conf/interpreter.json里有没有 flink 的配置项再看FLINK_HOME是否指向正确最后看 Zeppelin 日志里有没有加载 flink interpreter 时的报错。最常见的原因是 Flink 的 jar 和 Zeppelin 的 Scala 版本不匹配。问题二Remote 模式下提交任务报连接超时。检查zeppelin.flink.jobmanager.host和port是否可达。在 Zeppelin 所在机器上用telnet或者nc测一下 JobManager 的 RPC 端口。如果网络通但还报错可能是 Flink 集群的jobmanager.rpc.address配置和实际监听地址不一致。问题三SQL 能解析但执行时报ClassNotFoundException。这是连接器 jar 没加载。确认对应的连接器 jar 已经放到了 Zeppelin 能加载的路径下并且在 Interpreter 配置的flink.execution.jars里引用了。改完配置记得重启 Interpreter。4.2 SQL 执行类问题问题四流式聚合查询报“需要更新模式”。前面提过加typeupdate参数即可。根本原因是 Flink 的聚合查询产生的是更新流retract stream而 Zeppelin 默认按追加流展示两者不匹配。问题五时间戳字段显示成乱码或者时区不对。Flink SQL 的TIMESTAMP类型默认用 UTC 时区。如果数据里的时间是本地时间展示出来会差几个小时。解决办法是在建表时用TIMESTAMP_LTZ类型或者在 Zeppelin 的 Flink 配置里设置table.local-time-zone参数。问题六任务提交后一直处于 INITIALIZING 状态。多半是资源不够。检查 Flink 集群的 TaskManager 是否有足够的 slot。Remote 模式下Zeppelin 提交的任务和普通 Flink 任务一样占用 slotslot 满了就得等。我把常见问题整理成一张速查表方便对照现象可能原因解决方向看不到 flink 解释器Interpreter 未加载检查 FLINK_HOME 和 jar连接 JobManager 超时地址/端口错误测网络连通性ClassNotFoundException连接器 jar 缺失补 jar 并重启 Interpreter聚合查询报错展示模式不对加 typeupdate时间显示偏差时区配置设 local-time-zone任务卡在初始化资源不足检查 TaskManager slot4.3 几个独家避坑技巧第一个技巧善用%flink.conf段落。Zeppelin 允许在笔记本里用%flink.conf动态设置 Flink 配置比如并行度、Checkpoint 间隔等。这样不同笔记本可以用不同配置不用改全局的 Interpreter 设置。第二个技巧把常用的建表语句存成模板。Flink SQL 的建表语句又长又容易写错我习惯把 Kafka、JDBC、Hive 的建表模板存下来用的时候改改参数就行省时省力还不容易出错。第三个技巧调试阶段用 datagen 连接器造数据。不要一上来就连生产 Kafka用datagen造点模拟数据把 SQL 逻辑跑通确认无误后再换成真实数据源。这样既安全又高效。第四个技巧注意段落之间的变量传递。在%flink段落里定义的变量可以在后续%flink.ssql段落里通过${变量名}引用。这个特性在做参数化查询时很有用但要注意变量作用域跨笔记本是不共享的。5. 从调试到生产的进阶用法5.1 用 Zeppelin 管理 Flink 作业生命周期Zeppelin 不只是个 SQL 编辑器它还能管理 Flink 作业的生命周期。在%flink段落里可以通过 Flink 的 API 拿到执行环境然后做更细粒度的控制。比如设置 Checkpoint%flink val env StreamExecutionEnvironment.getExecutionEnvironment env.enableCheckpointing(60000) env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)或者在 SQL 段落里通过SET语句设置%flink.ssql SET execution.checkpointing.interval 60s; SET execution.checkpointing.mode EXACTLY_ONCE;这些配置对生产作业的稳定性至关重要。调试阶段可以关掉 Checkpoint 加快速度上线前一定要打开。5.2 多引擎协作与数据流转Zeppelin 的一个隐藏优势是多解释器协作。同一个笔记本里你可以先用%flink.ssql把 Kafka 数据加工后写入一张结果表再用%jdbc或者%hive段落去查这张表做验证。这种“流处理 批查询”的组合在实时数仓的日常运维里非常实用。具体做法是Flink SQL 把结果写到 Hive 表或者 JDBC 表然后在另一个段落里用对应的解释器查询。Zeppelin 会自动管理这些解释器的连接你只需要在段落开头切换%前缀即可。5.3 权限与多人协作生产环境用 Zeppelin绕不开权限问题。Zeppelin 支持基于 Shiro 的认证和授权可以配置用户名密码登录也可以对接 LDAP。配置在conf/shiro.ini里改完重启生效。多人协作方面Zeppelin 的笔记本可以共享给其他用户支持只读和可编辑两种权限。团队里我一般建议公共的建表模板和工具段落设成只读个人的调试笔记本各自维护。这样既保证了规范统一又不影响个人效率。提示Zeppelin 的笔记本数据默认存在本地文件系统notebook目录。如果要多人共享建议配置成存到数据库或者共享存储否则换台机器就找不到了。6. 一些实际使用中的体会这套环境我从早期版本一路用到现在最大的感受是它把 Flink SQL 的开发门槛实实在在地降下来了。以前调一个 SQL 要打包、提交、看日志一轮下来十几分钟现在改完直接跑几秒钟出结果。对于需要反复试错的实时计算场景这个效率提升是数量级的。但也要清醒地认识到它的边界。Zeppelin 适合开发和调试不适合直接当生产调度平台用。生产作业的提交、监控、告警还是得靠专门的调度系统。我一般把 Zeppelin 定位成“SQL 的试验台”逻辑在这里跑通、验证无误后再固化到调度平台上去。最后分享一个小技巧Zeppelin 的笔记本可以导出成 JSON 文件我习惯把每个项目的核心 SQL 笔记本导出备份。这样换环境或者重装的时候直接导入就能恢复比重新写一遍省事得多。另外定期清理不再使用的笔记本和解释器实例能避免 Zeppelin 越用越慢——这个坑我踩过笔记本多了之后启动明显变卡清理完就顺畅了。