企业官网建设流程全解析

1. 为什么要在 Zeppelin 里跑 Flink SQL

第一次接触 Flink SQL on Zeppelin 的人，脑子里通常有个疑问：Flink 自带的 SQL Client 已经能提交作业了，为什么还要套一层 Zeppelin？我当初也是这么想的，直到在一个多数据源探查的场景里被现实教育了一顿。当时需要在同一个会话里反复切换 Kafka 源表、MySQL 维表、Hive 结果表，还要边写边看结果，SQL Client 每次都要重新建表、重新连接，改一个字段就得重来一遍，效率低得让人抓狂。

Zeppelin 解决的核心问题就是交互式探索。它把 Flink SQL 变成了 Notebook 里的一个段落，你可以一段一段地写、一段一段地跑，中间结果直接以表格形式渲染出来，不用去翻日志、不用去猜。对于做数据探查、临时分析、SQL 调试的人来说，这种体验是质变。它适合三类人：一是数据开发，需要快速验证一段逻辑对不对；二是数据分析，不想写完整的 Flink 作业就想看看数据长什么样；三是运维和排查，线上作业出问题了，想用同样的 SQL 在交互环境里复现一下。

这篇文章我会把 Flink SQL on Zeppelin 的安装、配置、使用、踩坑完整讲一遍。安装部分会覆盖两种主流方式，配置部分会重点讲解释器参数怎么调，使用部分会给出可直接抄的 SQL 示例，最后会把我自己踩过的坑整理成速查表。整个过程基于 Flink 1.13 到 1.17 这几个版本，Zeppelin 用的是 0.10 和 0.11，这两个组合是社区里验证比较充分的。

需要提前说明的是，Flink 和 Zeppelin 的版本匹配非常关键，版本对不上会出现各种莫名其妙的类冲突。我下面给出的组合都是实测能跑通的，如果你用的是其他版本，思路一样，但具体路径和参数名可能需要微调。

2. 安装前的环境准备与版本选型

2.1 版本匹配是第一个大坑

Flink 和 Zeppelin 的集成是通过 Zeppelin 的 Flink 解释器实现的，这个解释器本质上是一个 Flink 客户端，它需要和 Flink 集群的版本严格对应。我见过太多人随便下两个版本就往一起凑，结果启动就报NoSuchMethodError或者ClassNotFoundException，排查半天发现是版本不匹配。

下面这张表是我实测过的组合，你可以直接参考：

选型建议很直接：如果你是新搭环境，直接上 Flink 1.15 + Zeppelin 0.11.0，这个组合社区文档最全，踩坑最少。如果你已经有现成的 Flink 集群，那就以集群版本为准，去选对应的 Zeppelin 版本，不要反过来。

2.2 基础环境依赖

在装之前，先把这些基础依赖确认一遍，缺一个都可能卡住：

• JDK：Flink 1.15 之前用 JDK 8，1.15 之后官方推荐 JDK 11。Zeppelin 对 JDK 11 的支持在 0.11 才比较完善，所以如果你用 JDK 11，Zeppelin 必须上 0.11。
• Scala：Flink 的二进制包分 Scala 2.11 和 2.12 两个版本，Zeppelin 的解释器也要对应。现在主流是 2.12，除非你有历史包袱，否则选 2.12。
• Hadoop：如果你的 Flink 要读写 HDFS 或 Hive，需要带上对应的 Hadoop 依赖包。Zeppelin 本身不强制依赖 Hadoop，但解释器加载 Flink 时会用到。
• 内存：Zeppelin 服务端建议至少 2GB 堆内存，Flink 解释器每个会话还会额外占用，机器内存小于 8GB 的话跑起来会比较吃力。

我个人的习惯是先用java -version和echo $JAVA_HOME确认 JDK，然后free -h看内存，最后确认 Flink 集群的版本号。这三步做完再动手，能省掉后面一半的麻烦。

2.3 两种安装路径的选择

Zeppelin 的安装有两种方式：一种是直接用官方二进制包解压，另一种是从源码编译。99% 的场景用二进制包就够了，源码编译只在你要改解释器代码或者官方包缺依赖时才需要。

二进制包又分两种：带-bin-all后缀的包含所有解释器，带-bin-netinst后缀的是最小安装，需要联网下载解释器。我强烈建议用-bin-all，因为 Flink 解释器在 netinst 模式下下载经常因为网络问题失败，而且下载的解释器版本可能和你预期的不一致。

下载地址在 Zeppelin 官网的下载页，选对应版本即可。下载完直接解压，不需要编译，这是最省事的路子。

3. Zeppelin 的安装与 Flink 解释器配置

3.1 解压与目录结构说明

假设你下载的是zeppelin-0.11.0-bin-all.tgz，放到/opt下解压：

tar -zxvf zeppelin-0.11.0-bin-all.tgz -C /opt cd /opt/zeppelin-0.11.0-bin-all

解压后的目录结构里，你需要重点关注这几个：

• conf/：配置文件目录，zeppelin-site.xml和zeppelin-env.sh都在这里。
• interpreter/：解释器目录，Flink 解释器相关的 jar 包在这里。
• bin/：启动脚本，zeppelin-daemon.sh是核心。
• logs/：日志目录，出问题第一个来看这里。

先别急着启动，配置没弄好启动也是白搭。

3.2 配置 zeppelin-env.sh

从模板复制一份配置文件：

cp conf/zeppelin-env.sh.template conf/zeppelin-env.sh

然后编辑conf/zeppelin-env.sh，加上这几行：

export JAVA_HOME=/usr/lib/jvm/java-11-openjdk export ZEPPELIN_ADDR=0.0.0.0 export ZEPPELIN_PORT=8080 export ZEPPELIN_MEM="-Xmx2048m -XX:MaxPermSize=512m" export ZEPPELIN_JAVA_OPTS="-Dfile.encoding=UTF-8"

这里有几个点要解释一下。ZEPPELIN_ADDR设成0.0.0.0是为了让其他机器也能访问，如果你只在本机用，设成127.0.0.1更安全。ZEPPELIN_MEM是服务端内存，2GB 是起步值，如果你的会话多，要往上加。ZEPPELIN_JAVA_OPTS里的编码设置很重要，不设的话中文注释和中文数据可能乱码。

3.3 配置 Flink 解释器

这是最关键的一步。Zeppelin 的 Flink 解释器配置有两种方式：一种是在 Web UI 里改，另一种是直接改配置文件。我推荐直接改配置文件，因为 Web UI 改完重启可能丢，配置文件更可靠。

编辑conf/zeppelin-site.xml，找到 Flink 解释器相关的配置项。如果没有，就手动加上：

<property> <name>zeppelin.interpreter.flink.flink.home</name> <value>/opt/flink-1.15.0</value> </property> <property> <name>zeppelin.interpreter.flink.flink.conf.dir</name> <value>/opt/flink-1.15.0/conf</value> </property>

flink.home指向你的 Flink 安装目录，flink.conf.dir指向 Flink 的配置目录。这两个路径必须写对，写错了解释器启动时会报找不到 Flink 配置。

还有一个隐藏的坑：Zeppelin 的 Flink 解释器默认用的是它自己interpreter/flink/目录下的 jar 包，而不是你 Flink 安装目录下的。如果版本不一致，就会出现类冲突。解决办法是把 Flink 安装目录下lib/里的核心 jar 包复制到 Zeppelin 的interpreter/flink/目录，覆盖掉原来的。具体要复制哪些，我一般是把flink-dist、flink-table、flink-sql相关的都拷过去。

3.4 启动与验证

配置改完，启动 Zeppelin：

bin/zeppelin-daemon.sh start

启动后看日志确认没报错：

tail -f logs/zeppelin-*.log

日志里出现ZeppelinServer started就说明服务起来了。然后浏览器访问http://你的机器IP:8080，能看到 Zeppelin 首页就成功了。

接下来验证 Flink 解释器。在首页点Create new note，Note Name 随便填，Default Interpreter 选flink。创建后进入 Notebook，输入一段最简单的 SQL：

%flink SELECT 1 + 1

按 Shift+Enter 执行，如果下方输出一个表格显示2，说明解释器工作正常。如果报错，去看日志里 Flink 解释器相关的部分，通常是版本不匹配或者路径配错。

注意：第一次执行 Flink SQL 会比较慢，因为解释器要初始化 Flink 客户端和连接集群，等个十几秒是正常的，不要以为卡死了。

4. Flink SQL on Zeppelin 的核心用法

4.1 解释器模式的选择

Zeppelin 的 Flink 解释器支持几种模式，这个在创建 Note 或者配置解释器时可以选择。不同模式决定了你的 SQL 跑在什么环境里：

• local 模式：Flink 以本地进程方式运行，不连集群。适合快速验证语法和逻辑，数据量小的时候用。
• remote 模式：连到已有的 Flink 集群，作业提交到集群上跑。适合生产环境调试。
• application 模式：每个会话启动一个独立的 Flink 应用，资源隔离好，但启动慢。

我日常用得最多的是 local 模式做语法验证，remote 模式做真实数据调试。切换模式在解释器配置里改zeppelin.interpreter.flink.mode这个参数，或者在 Note 的顶部用%flink.conf段落动态设置。

4.2 用 %flink 段落写 SQL

Zeppelin 里 Flink SQL 的基本写法是在段落开头加%flink，然后直接写 SQL。比如建一张 Kafka 源表：

%flink CREATE TABLE kafka_source ( user_id BIGINT, item_id BIGINT, behavior STRING, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'user_behavior', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'zeppelin_group', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' )

执行后如果没报错，表就建好了。这里有个细节：Zeppelin 的 Flink 解释器默认是流模式，建的表是流表。如果你想跑批，需要在段落里显式设置执行模式：

%flink.conf execution.runtime-mode batch

然后再写批 SQL。这个配置是会话级的，设置后当前 Note 后续的段落都按批模式跑。

4.3 查询结果的可视化

Zeppelin 最爽的一点是查询结果能直接可视化。比如你跑一个聚合查询：

%flink SELECT behavior, COUNT(*) AS cnt FROM kafka_source GROUP BY behavior

结果会以表格形式展示。点击表格上方的图表图标，可以切换成柱状图、饼图、折线图等。对于流查询，Zeppelin 会持续刷新结果，你能看到数据实时变化。

不过要注意，流查询默认是无限跑的，Zeppelin 会一直占用资源。如果你只想看一段时间的数据，可以在 SQL 里加时间窗口限制，或者手动点停止按钮。我一般会在测试时加个WHERE ts > TIMESTAMP '2024-01-01 00:00:00'之类的条件，避免全量扫描。

4.4 多段落协作的写法

Zeppelin 的 Note 里可以有多个段落，段落之间共享同一个 Flink 会话。这意味着你可以在第一个段落建表，第二个段落插入数据，第三个段落查询，它们用的是同一个环境。

这种写法特别适合做数据探查。比如：

段落一建源表，段落二建结果表，段落三写 INSERT INTO 把数据从源表导到结果表，段落四查询结果表验证。每个段落单独执行，出错了只重跑出错的那段，不用从头来。

我常用的一个模式是：把建表语句放在最上面的段落，用%flink.conf设置好并行度和 checkpoint 间隔，然后下面每个段落专注做一件事。这样调试的时候思路清晰，也方便把验证好的 SQL 直接搬到生产作业里。

5. 实操案例：从 Kafka 到 MySQL 的完整链路

5.1 场景说明与表设计

光讲语法没意思，我拿一个真实做过的场景来串一遍。需求是这样的：从 Kafka 读用户行为数据，实时统计每个商品的点击量，把结果写入 MySQL 供报表查询。

数据流是：Kafka（JSON 格式行为日志）-> Flink SQL 聚合 -> MySQL 结果表。

先在 Zeppelin 里建 Kafka 源表。假设 Kafka 里的消息格式是：

{"user_id": 1001, "item_id": 2001, "behavior": "click", "ts": "2024-01-15 10:30:00"}

对应的建表语句：

%flink CREATE TABLE user_behavior ( user_id BIGINT, item_id BIGINT, behavior STRING, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'user_behavior', 'properties.bootstrap.servers' = 'kafka-broker:9092', 'properties.group.id' = 'zeppelin_behavior_group', 'scan.startup.mode' = 'latest-offset', 'format' = 'json', 'json.fail-on-missing-field' = 'false', 'json.ignore-parse-errors' = 'true' )

这里json.fail-on-missing-field和json.ignore-parse-errors两个参数很关键。生产环境的 JSON 数据经常有字段缺失或格式错误，不设这两个参数，一条脏数据就能让整个作业挂掉。设成 false 和 true 后，脏数据会被跳过，作业能继续跑。

5.2 聚合逻辑与结果表

接下来建 MySQL 结果表：

%flink CREATE TABLE item_click_count ( item_id BIGINT, click_count BIGINT, window_end TIMESTAMP(3), PRIMARY KEY (item_id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://mysql-host:3306/report', 'table-name' = 'item_click_count', 'username' = 'report_user', 'password' = 'your_password', 'sink.buffer-flush.max-rows' = '100', 'sink.buffer-flush.interval' = '5s' )

MySQL 表需要提前在数据库里建好，Flink 不会自动建表。建表 SQL 大概是：

CREATE TABLE item_click_count ( item_id BIGINT PRIMARY KEY, click_count BIGINT, window_end DATETIME );

然后写聚合和写入逻辑：

%flink INSERT INTO item_click_count SELECT item_id, COUNT(*) AS click_count, TUMBLE_END(ts, INTERVAL '1' MINUTE) AS window_end FROM user_behavior WHERE behavior = 'click' GROUP BY item_id, TUMBLE(ts, INTERVAL '1' MINUTE)

这段 SQL 用的是滚动窗口，每分钟统计一次。TUMBLE_END给出窗口结束时间，方便报表按时间查询。

5.3 参数调优与资源控制

在 Zeppelin 里跑流作业，资源控制很重要，不然一个会话能把集群资源吃光。我一般会在 Note 开头加一个配置段落：

%flink.conf parallelism.default 2 execution.checkpointing.interval 60000 execution.checkpointing.mode EXACTLY_ONCE state.backend rocksdb state.backend.incremental true

parallelism.default设成 2 是因为测试环境资源有限，生产上按数据量调。checkpoint 间隔 60 秒是折中值，太短影响性能，太长故障恢复慢。state.backend用 rocksdb 是因为聚合状态可能比较大，内存存不下。

还有一个容易忽略的点：Zeppelin 的 Flink 解释器默认会为每个段落创建一个新的 Job，如果你在同一个 Note 里跑了多个 INSERT 语句，它们会变成多个独立的 Flink 作业。这在调试时没问题，但如果你想模拟一个完整的作业，应该把所有逻辑写在一个段落里，或者用 Statement Set 把多个 INSERT 合并。

5.4 验证与结果查看

作业跑起来后，怎么确认数据写进去了？两个办法。

一是在 Zeppelin 里直接查 MySQL 结果表：

%flink SELECT * FROM item_click_count ORDER BY window_end DESC LIMIT 10

二是去 MySQL 客户端里查，这个更直接。如果 Zeppelin 里查不到数据，但 MySQL 里有，说明是 Zeppelin 的查询有问题；如果两边都没有，那就是作业没跑起来或者数据没进来。

我排查这类问题时，习惯先在 Zeppelin 里跑一个SELECT * FROM user_behavior LIMIT 10，确认源数据能读到。源数据没问题，再查聚合逻辑。这个顺序能快速定位问题出在哪一环。

6. 常见问题与排查技巧实录

6.1 解释器启动失败类问题

这是最常见的一类问题，表现是创建 Note 后执行任何 SQL 都报错，或者解释器状态显示红色。

问题一：ClassNotFoundException 或 NoSuchMethodError

原因几乎都是版本不匹配。Flink 解释器加载的 jar 包版本和 Flink 集群版本不一致。解决办法是确认 Zeppelin 的interpreter/flink/目录下的 jar 包版本，和 Flink 安装目录lib/下的版本一致。不一致就把 Flink 的拷过去覆盖。

问题二：找不到 Flink 配置

报错信息里出现Could not find flink-conf.yaml之类的。检查zeppelin-site.xml里的flink.conf.dir路径对不对，以及该路径下确实有flink-conf.yaml文件。

问题三：端口占用或连接拒绝

如果用的是 remote 模式，报连接拒绝，先确认 Flink 集群的 JobManager 地址和端口配对了，再确认网络能通。用telnet jobmanager-host 8081测一下。

6.2 SQL 执行报错类问题

问题一：表已存在

Zeppelin 的会话是持久的，你建过的表会一直在。重复执行建表语句会报Table already exists。解决办法是在建表前加DROP TABLE IF EXISTS xxx，或者用CREATE TABLE IF NOT EXISTS。

问题二：时间属性字段报错

Flink SQL 对时间字段要求严格，TIMESTAMP(3)和TIMESTAMP是两回事。Kafka 里的时间字段如果是字符串，需要显式转换。我一般会在建表时用计算列：

ts AS TO_TIMESTAMP(json_ts)

问题三：连接器找不到

报Could not find any factory for identifier 'kafka'，说明 Flink 的 Kafka 连接器 jar 包没加载。需要把flink-connector-kafka的 jar 包放到 Flink 的lib/目录，或者放到 Zeppelin 解释器的 classpath 里。

6.3 性能与资源类问题

问题一：查询一直不返回

流查询默认无限跑，如果你写的是SELECT * FROM kafka_source，它会一直等新数据。加个 LIMIT 或者时间范围限制，或者手动停止。

问题二：内存溢出

报OutOfMemoryError，通常是状态太大或者并行度太高。调小并行度，或者换 rocksdb 状态后端，或者加内存。

问题三：Checkpoint 一直失败

看日志里 checkpoint 失败的原因，常见的是状态太大超时，或者存储路径不可写。调大 checkpoint 超时时间，确认存储路径权限。

下面这张表是我整理的速查表，遇到问题先对号入座：

6.4 几个独家避坑技巧

第一个技巧：建表语句和查询语句分段落放。建表放一个段落，查询放另一个。这样改查询的时候不用重跑建表，改表结构的时候也只影响建表段落。我见过有人把所有 SQL 写在一个段落里，改一个字段要全部重跑，效率极低。

第二个技巧：用 %flink.conf 段落统一管理配置。把并行度、checkpoint、状态后端这些配置集中在一个段落里，不要散落在各个 SQL 段落。这样换环境的时候只改一个地方。

第三个技巧：善用 Zeppelin 的版本管理。Zeppelin 支持 Note 的版本历史，每次修改都会记录。调 SQL 的时候如果改乱了，可以回滚到之前的版本。这个功能在调试复杂逻辑时特别有用。

第四个技巧：流作业调试时先用 local 模式。local 模式不连集群，启动快，验证语法和逻辑足够了。确认没问题再切 remote 模式跑真实数据。这样能省掉大量等待集群调度的时间。

7. 从 Zeppelin 到生产作业的迁移

在 Zeppelin 里调通的 SQL，最终是要搬到生产作业里的。这个迁移过程有几个注意点。

首先是配置的差异。Zeppelin 里的%flink.conf配置是会话级的，搬到生产作业时要转成 Flink 的配置文件或者代码里的配置。比如并行度、checkpoint 间隔这些，在生产作业里通常通过flink-conf.yaml或者StreamExecutionEnvironment设置。

其次是连接器的依赖。Zeppelin 里能跑通，是因为解释器的 classpath 里有对应的连接器 jar。生产作业打包时，要把这些连接器依赖打进去，或者放到 Flink 的lib/目录。我一般用 Maven 的 shade 插件把依赖打成一个 fat jar。

然后是SQL 的兼容性。Zeppelin 里用的 Flink 版本和生产集群的版本可能不完全一致，一些新语法在旧版本上不支持。迁移前先在目标版本上验证一遍。

最后是状态的兼容性。如果生产作业之前有状态，新作业的 SQL 改了逻辑，可能导致状态不兼容。这种情况要么用 savepoint 做状态迁移，要么接受状态丢失重新计算。

我个人的做法是：在 Zeppelin 里调通逻辑后，把 SQL 复制到一个独立的.sql文件里，用 Flink SQL Client 或者sql-submit脚本提交到集群验证一遍，确认没问题再集成到正式的作业代码里。这样多一道验证，能避免很多低级错误。

Zeppelin 的定位始终是交互式探索工具，不是生产调度工具。它的价值在于让你快速看到数据、快速验证想法，而不是替代正式的作业管理。把这两者的边界分清楚，用起来就顺手了。