Flink 2.2 Docker 部署Session / Application / SQL Client 一把梭（含 Compose、插件、连接器与踩坑点）

1. 先搞清楚：Docker 里跑 Flink，你到底在跑什么

Flink 的官方镜像本质上就是一个标准 Flink 发行版 + 入口脚本。你可以用它启动三种角色：

• JobManager（Session 集群）
• JobManager（Application 集群，standalone-job）
• TaskManager（给任意集群用）

核心差别只在于：你是先起集群再提交作业（Session），还是“集群=作业”（Application）。

2. 最快启动：手动 docker run 启一个 Session 集群

2.1 创建网络 + 指定 JobManager RPC 地址

JobManager 和 TaskManager 要互相发现，所以必须设置jobmanager.rpc.address，并且建议放进同一个 Docker network。

FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager"dockernetwork create flink-network

2.2 启 JobManager（带 Web UI）

dockerrun\--rm\--name=jobmanager\--network flink-network\--publish8081:8081\--envFLINK_PROPERTIES="${FLINK_PROPERTIES}"\flink:2.2.0-scala_2.12 jobmanager

2.3 启 TaskManager（可多个）

dockerrun\--rm\--name=taskmanager\--network flink-network\--envFLINK_PROPERTIES="${FLINK_PROPERTIES}"\flink:2.2.0-scala_2.12 taskmanager

2.4 提交作业（在你本机有 Flink 客户端的前提下）

./bin/flink run ./examples/streaming/TopSpeedWindowing.jar

访问 Web UI：http://localhost:8081

停止集群：直接 Ctrl+C 或docker stop jobmanager taskmanager

3. 推荐方式：Docker Compose 一键起 Session / Application / SQL Client

Compose 的优势是：配置集中、扩缩容方便、可复制到其他机器。

3.1 Session Mode（最常用）

docker-compose.yml示例：

version:"2.2"services:jobmanager:image:flink:2.2.0-scala_2.12ports:-"8081:8081"command:jobmanagerenvironment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanagertaskmanager:image:flink:2.2.0-scala_2.12depends_on:-jobmanagercommand:taskmanagerscale:1environment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanager taskmanager.numberOfTaskSlots: 2

启动：

dockercompose up

扩容 TaskManager（两种写法二选一）：

dockercompose scaletaskmanager=3# 或者新版本：docker compose up --scale taskmanager=3

关闭：

dockercompose down

3.2 SQL Client + Session 集群（本地写 SQL 最爽的组合）

version:"2.2"services:jobmanager:image:flink:2.2.0-scala_2.12ports:-"8081:8081"command:jobmanagerenvironment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanagertaskmanager:image:flink:2.2.0-scala_2.12depends_on:-jobmanagercommand:taskmanagerscale:1environment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanager taskmanager.numberOfTaskSlots: 2sql-client:image:flink:2.2.0-scala_2.12command:bin/sql-client.shdepends_on:-jobmanagerenvironment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanager rest.address: jobmanager

启动集群后进入 SQL Client：

dockercompose run sql-client

重要提醒：SQL Client 只是“提交端”，真正跑算子的还是 JM/TM。你用到的连接器依赖必须在集群侧也可见（见第 5 节）。

3.3 Application Mode（一个应用一个集群，隔离更强）

Application Mode 的关键点：作业 JAR 必须在容器里可用（通常/opt/flink/usrlib），或者通过--jars指定远端下载路径。

Compose 示例（挂载 usrlib）：

version:"2.2"services:jobmanager:image:flink:2.2.0-scala_2.12ports:-"8081:8081"command:>standalone-job --job-classname com.job.ClassNamevolumes:-/host/path/to/job/artifacts:/opt/flink/usrlibenvironment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanager parallelism.default: 2taskmanager:image:flink:2.2.0-scala_2.12depends_on:-jobmanagercommand:taskmanagerscale:1volumes:-/host/path/to/job/artifacts:/opt/flink/usrlibenvironment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanager taskmanager.numberOfTaskSlots: 2 parallelism.default: 2

如果你要从 savepoint 拉起（Application Mode 常见需求）：

• --fromSavepoint /path/to/savepoint
• --allowNonRestoredState（可选）

注意：savepoint 路径必须在所有容器里都能访问（DFS 或挂载卷）。

4. 三种配置方式：动态参数、环境变量、挂载 conf

你可以用任意一种“覆盖”容器内的flink-conf.yaml。

4.1 动态参数（-D）覆盖

dockerrun flink:2.2.0-scala_2.12 taskmanager\-D jobmanager.rpc.address=jobmanager\-D taskmanager.numberOfTaskSlots=3\-D blob.server.port=6124

4.2 环境变量 FLINK_PROPERTIES（最常用）

FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager taskmanager.numberOfTaskSlots: 3 blob.server.port: 6124 "dockerrun --envFLINK_PROPERTIES="${FLINK_PROPERTIES}"flink:2.2.0-scala_2.12 taskmanager

提示：jobmanager.rpc.address必配，其它按需。

4.3 挂载自定义 conf（你要全量接管配置）

dockerrun\--mounttype=bind,src=/host/path/to/custom/conf,target=/opt/flink/conf\flink:2.2.0-scala_2.12 jobmanager

注意：挂载目录里要包含必要的配置文件，并且 Flink 配置文件要可写（入口脚本有时会修改）。

5. 连接器与 SQL Client：最容易踩的坑（也是最关键的坑）

坑 1：连接器 JAR 只放在 SQL Client 里不行

SQL Client 只是提交 SQL，真正执行 SQL 的运行时在 JM/TM。你需要确保：

• JobManager / TaskManager 的/opt/flink/lib（或 plugin 路径）里也有相同连接器 JAR
• SQL Client 容器里也有（否则解析/提交阶段可能缺类）

坑 2：ADD JAR对宿主机路径不生效

在容器里，“本地文件系统”是 Docker overlay filesystem，不是你的宿主机路径。想让集群看到 JAR：

• 方案 A：自定义镜像，把 JAR 放进/opt/flink/lib
• 方案 B：挂载卷，把 JAR 挂进/opt/flink/lib或/opt/flink/usrlib
• 方案 C：Application Mode 用--jars从 S3/HTTP/DFS 拉

推荐：做一个带 Kafka 连接器的自定义镜像

示例 Dockerfile（官方示例思路）：

FROM flink:2.2.0-scala_2.12 ARG kafka_connector_version=4.0.0-2.0 RUN wget -P /opt/flink/lib https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/${kafka_connector_version}/flink-sql-connector-kafka-${kafka_connector_version}.jar

然后在 compose 里build替换image，让 jobmanager/taskmanager/sql-client 都用同一镜像。

6. 插件启用：S3、Hadoop FS 这些“不是放 lib 就完事”

Flink 的 filesystem 插件通常需要放到/opt/flink/plugins/<plugin-name>/下，官方镜像提供了ENABLE_BUILT_IN_PLUGINS来启用自带插件 JAR（从/opt/flink/opt链接到 plugins）。

示例：

dockerrun\--envENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-2.2.0.jar\flink:2.2.0-scala_2.12 jobmanager

多个插件用;分隔。

7. 容器里做“本地恢复”：Working Directory + 挂载卷（强烈建议）

如果你用了 RocksDB state backend 或者希望 TaskManager 重启后更快恢复，建议把 working dir 做成“可持久化”挂载卷，否则容器一重启，本地 state 直接消失。

建议做法：

• 挂载一个宿主机目录到容器，例如/data/flink/working -> /data/flink/working
• 配置process.working-dir指向这个路径
• 开启state.backend.local-recovery: true
• 给 TaskManager 固定taskmanager.resource-id（在容器环境尤其重要）

Compose 片段示例（思路）：

services:taskmanager:image:flink:2.2.0-scala_2.12volumes:-/data/flink/working:/data/flink/workingenvironment:-|FLINK_PROPERTIES= jobmanager.rpc.address: jobmanager process.working-dir: /data/flink/working state.backend.local-recovery: true taskmanager.resource-id: tm-01

如果你会 scale 多个 TM，要让每个实例的taskmanager.resource-id唯一，否则会踩目录冲突坑。生产里更常见的做法是“每台机器固定一个 TM + 固定 resource-id”，或用编排系统注入唯一 ID。

8. 内存分配器：jemalloc 与 glibc 的选择

官方镜像默认使用 jemalloc 来缓解内存碎片问题。如果你想切回 glibc：

dockerrun --envDISABLE_JEMALLOC=true flink:2.2.0-scala_2.12 taskmanager

如果你使用 glibc，建议限制 arena 数避免内存膨胀（尤其是 RocksDB + checkpoint/savepoint 时更明显）：

dockerrun --envMALLOC_ARENA_MAX=1flink:2.2.0-scala_2.12 taskmanager

9. 在 Docker 里跑 PyFlink：自定义镜像一步到位

示例 Dockerfile：

FROM flink:2.2.0 RUN apt-get update -y && \ apt-get install -y python3 python3-pip python3-dev && \ rm -rf /var/lib/apt/lists/* RUN ln -s /usr/bin/python3 /usr/bin/python RUN pip3 install apache-flink==2.2.0

构建：

dockerbuild -t pyflink:latest.

10. 镜像标签最佳实践：别用 latest，别偷懒

强烈建议显式使用包含 Flink+Scala 的 tag，比如：

• flink:2.2.0-scala_2.12

原因很现实：Scala 版本与 Flink 版本不匹配时，类冲突会让你怀疑人生。

官方镜像有两个渠道：

• Docker Hub 官方flink（更推荐）
• apache/flink（社区维护，适合应急）

11. 一套“最短闭环”的调试套路（特别适合你做压测/验证）

你前面提到 “Print 验证正确性 + BlackHole 压测性能”，在 Docker Session 集群里尤其好用：

• 开发阶段：source 用datagen，sink 用print看输出是否符合预期
• 压测阶段：sink 换blackhole，你就能把瓶颈收敛到 join/agg/topn/UDF 本身（避免 sink 抢戏）
• 观察方式：看 Web UI 的 backpressure、busy、checkpoint（如果有）、以及算子吞吐