【免费下载链接】flink
项目地址：
https://gitcode.com/gh_mirrors/fli/flink
点击查看 免费下载
本文导读：本文以 Flink 官方部署文档《配置参数》为骨架，系统讲解 Apache Flink 从 1.19 版本起引入的 config.yaml 标准 YAML 配置体系——包括嵌套/扁平两种写法、YAML 1.2 值类型支持、与旧版 flink-conf.yaml 的行为差异及迁移工具的使用，并在此基础上深入解析 Basic Setup、Common Setup、Security、资源编排、状态后端、Metrics、Execution 等核心配置域，结合本仓库源码给出可直接落地的配置示例与底层原理。读完本文，你将掌握 Flink 配置文件的完整读写方法、一键迁移流程，以及围绕内存、并行度、Checkpoint、高可用等场景的实战调优能力。
配置文件体系总览
Flink 的所有配置都可以写在 conf/ 目录下的配置文件中（参见 Flink 配置文件一节）。配置在 Flink 进程启动时被解析和加载，修改配置文件后必须重启相关进程才能生效。
默认配置直接使用系统当前的 Java 安装。若需手动指定 Java 运行时，可以通过环境变量 JAVA_HOME，也可以在配置文件中设置 env.java.home 配置项。注意：env.java.home 必须以扁平格式（即单行 key-value 形式）书写。
可以通过定义 FLINK_CONF_DIR 环境变量来指定不同的配置目录。对于提供非会话（non-session）部署的资源提供者，你可以用这种方式为每个作业单独配置——即拷贝一份 conf 目录、按作业修改配置。但该方式不支持 Docker 或 standalone Kubernetes 部署；在 Docker 部署中，应使用 FLINK_PROPERTIES 环境变量传递配置值。
对于 Session 集群，提供的配置只会用于配置 Execution 相关参数（即影响作业的参数），而不会影响底层集群本身。
Flink 配置文件
自 Flink-1.19 起，Flink 正式引入对标准 YAML 1.2 语法的完整支持。相比旧版仅支持简单键值对的 flink-conf.yaml，新配置文件 config.yaml 提供更灵活强大的配置能力。原 flink-conf.yaml 不再推荐使用，并将在 Flink-2.0 中不再支持。建议尽早将现有作业配置迁移到新的配置方式。
用法
从 Flink-1.19 开始，默认配置文件已改为 conf/ 目录下的 config.yaml，用户直接修改该文件即可。
若希望继续使用旧文件，只需将 flink-conf.yaml 拷贝到 conf/ 目录。一旦检测到 flink-conf.yaml，Flink 会优先使用它作为配置文件。
config.yaml 的配置方式如下：
Config Key
Nested（嵌套）格式——按层级组织 Config Key：
restart-strategy:
type: failure-rate
failure-rate:
delay: 1 s
failure-rate-interval: 1 min
max-failures-per-interval: 1
Flatten（扁平）格式——用点号连接层级：
restart-strategy.type: failure-rate
restart-strategy.failure-rate.delay: 1 s
restart-strategy.failure-rate.failure-rate-interval: 1 min
restart-strategy.failure-rate.max-failures-per-interval: 1
两种写法等价，可以混用。仓库自带的 config.yaml 模板 即大量采用嵌套格式，例如 jobmanager.rpc.port、taskmanager.memory.process.size、parallelism.default 等。
Config Value
config.yaml 支持按 YAML 1.2 core schema 配置 value 类型，即可以根据值的内容自动识别类型（字符串、整数、浮点数、布尔、Null 等），无需像旧格式那样全部视为字符串。例如：
# 布尔值
web.submit.enable: true
# 整数
parallelism.default: 4
# 带单位的时间/内存（仍为字符串写法，由 ConfigOption 解析）
execution.checkpointing.interval: 3min
jobmanager.memory.process.size: 1600m
# 显式字符串
rest.address: "my-flink-jobmanager.example.com"
此外，还可以用单引号或双引号将原始值括起来，强制按字符串格式配置任意类型的 value。
从 flink-conf.yaml 迁移至 config.yaml
行为变更
config.yaml 严格遵循 YAML 1.2 语法，与 flink-conf.yaml 在大部分情况下完全兼容，但以下场景发生了行为变更：
| Null value | 仅支持留空 | 支持留空、null、Null、NULL 和 ~ |
| 注释 | 每行中首次出现 # 之后的内容都视为注释 | 仅当 # 前至少有一个空格、或 # 位于行首时，其后内容才视为注释 |
| 需要转义的特殊字符 | 仅需对 List 和 Map 元素转义（List 元素含 ;、Map 元素含 , 和 : 时需转义） | 需按 YAML 1.2 规范对 Indicator Characters 特殊字符转义 |
| 重复 Key | 允许重复，取文件中最末端出现的 key-value pair | 不允许重复，加载配置时直接报错 |
| 非法配置 | 非法的 key-value pair 被忽略 | 加载配置时报错 |
其中“非法配置直接报错”与“重复 Key 报错”意味着迁移后配置错误会显式暴露，而不再像旧版那样静默忽略，这有助于尽早发现问题。
Migration Tool
Flink 提供了官方迁移脚本，可自动化完成迁移：
bin/migrate-config-file.sh
该脚本的实现位于仓库 migrate-config-file.sh：它会读取 conf/ 下的旧配置，调用 bash-java-utils.sh 中的 migrateLegacyFlinkConfigToStandardYaml（见 bash-java-utils.sh），最终把迁移结果写入 conf/config.yaml。
由于旧解析器的限制，flink-conf.yaml 中所有 value 都会被识别为 String 类型，因此生成的 config.yaml 中部分 value 会被引号包裹；Flink 在后续配置解析时会将其转换为 ConfigOption 定义的实际类型，不影响最终生效值。
迁移完成后，必须删除 conf/ 目录下的 flink-conf.yaml，新的 config.yaml 才会生效。
Basic Setup（基础配置）
默认配置无需任何修改即可启动单节点 Flink Session 集群。本节参数是搭建基本分布式 Flink 集群最常用的配置。
Hostnames / Ports（主机名与端口）
以下选项仅对 standalone 的应用/会话部署（简单 standalone 或 Kubernetes standalone）是必需的。如果使用 Yarn 或 Native Kubernetes，主机名和端口会被自动发现。
- rest.address / rest.port：客户端连接 Flink 使用的地址与端口。设置为 JobManager 所在主机名，或 JobManager REST 接口前（Kubernetes）Service 的主机名。
- jobmanager.rpc.address（默认 localhost）与 jobmanager.rpc.port（默认 6123）：TaskManager 连接 JobManager/ResourceManager 使用的地址与端口。在高可用部署下该选项会被忽略，由 leader 选举机制自动发现。
Memory Sizes（内存大小）
默认内存足以运行简单的流/批应用，但对复杂应用偏小。
- jobmanager.memory.process.size：JobManager（JobMaster / ResourceManager / Dispatcher）进程总内存。
- taskmanager.memory.process.size：TaskManager 进程总内存。
总大小包含一切。Flink 会扣除 JVM 自身所需内存（metaspace 等），再在组件间自动分配剩余内存（JVM Heap、Off-Heap，TaskManager 还包括 Network、Managed Memory 等）。值按内存大小书写，例如 1536m 或 2g。
Parallelism（并行度）
- taskmanager.numberOfTaskSlots（默认 1）：每个 TaskManager 提供的 slot 数，每个 slot 可运行一个 task 或 pipeline。多 slot 有助于将 JVM、应用库、网络连接等固定开销分摊到多个并行任务上（详见 Task Slots and Resources）。
- 运行多个“单 slot 的小 TaskManager”是很好的起步方案，任务间隔离性最佳；把同样资源分配给“多 slot 的少量大 TaskManager”能提高资源利用率，但任务间隔离变弱（更多任务共享同一 JVM）。
- parallelism.default（默认 1）：未在任何地方指定并行度时的默认并行度。
Checkpointing（检查点）
这些值可在作业代码中直接配置；写在这里是作为应用未配置时的默认值。
- state.backend.type：使用的状态后端，决定快照的数据结构机制，常见值为 hashmap 或 rocksdb。
- execution.checkpointing.dir：检查点写入目录，接受 URI，如 s3://mybucket/flink-app/checkpoints 或 hdfs://namenode:port/flink/checkpoints。
- execution.checkpointing.savepoint-dir：Savepoint 默认目录，同样接受路径 URI。
- execution.checkpointing.interval：基础间隔设置，要启用检查点需将该值设为大于 0。
Web UI
- web.submit.enable（默认 true）：是否允许通过 Flink UI 上传并启动作业。注意即使关闭，Session 集群仍可通过 REST 请求（HTTP 调用）提交作业，该开关只保护 UI 上传功能。
- web.cancel.enable（默认 true）：是否允许通过 Flink UI 取消作业，关闭后仍可通过 REST 取消。
- web.upload.dir：上传作业的存储目录，仅当 web.submit.enable 为 true 时使用。
Other（其他）
- io.tmp.dirs：Flink 存放本地数据的目录，默认取系统临时目录（java.io.tmpdir）。配置多个目录时 Flink 会在目录间轮转文件。这些目录中的数据包括 RocksDB 创建的文件、溢写的中间结果（批处理算法）、缓存的 jar 文件等。
- 这些数据不依赖其做持久化/恢复，但若被删除通常会导致重量级恢复操作，因此建议设置到不会被定期自动清理的目录。
- Yarn 和 Kubernetes 部署默认会自动将其配置为本地工作目录。
Common Setup Options（常用配置）
配置 Flink 应用或集群的通用选项。
Hosts and Ports
各组件的主机名与端口配置。JobManager 主机名与端口仅对无高可用的 standalone 部署有意义——此时 TaskManager 靠这些值发现并连接 JobManager；在一切高可用部署中，TaskManager 通过高可用服务（如 ZooKeeper）发现 JobManager。使用资源编排框架（K8s、Yarn）的部署通常使用框架自身的服务发现机制。除非需要特定端口范围或绑定特定网卡，一般无需配置 TaskManager 主机与端口。
Fault Tolerance（容错）
以下选项控制执行失败时的重启行为。在 flink-conf.yaml 中配置即定义了集群默认重启策略；只有当作业未通过 ExecutionConfig 配置作业级重启策略时，默认策略才会生效。
内置三种重启策略：
Fixed Delay（固定延迟）
restart-strategy:
type: fixed-delay
fixed-delay:
attempts: 3
delay: 10 s
Exponential Delay（指数延迟）：重启延迟随尝试次数指数增长，适合长时间运行的流作业。
Failure Rate（失败率）
restart-strategy:
type: failure-rate
failure-rate:
delay: 1 s
failure-rate-interval: 1 min
max-failures-per-interval: 1
即 1 分钟内最多失败 1 次，超过则作业失败；否则等待 1 秒后重启。
Retryable Cleanup（可重试清理）
作业进入全局终态后，会清理所有相关资源。该清理失败时可重试，可通过不同重试策略改变行为：支持 Fixed-Delay Cleanup Retry Strategy 与 Exponential-Delay Cleanup Retry Strategy。
Checkpoints and State Backends
控制状态后端与检查点行为的基础选项。这些选项仅对连续流式作业相关；批处理作业不使用状态后端和检查点，而是使用针对批处理优化的内部数据结构。
High Availability（高可用）
这里的高可用指 JobManager 进程从故障中恢复的能力。JobManager 保证恢复期间跨 TaskManager 的一致性；为让 JobManager 自身一致恢复，外部服务需存储少量恢复元数据（如“最后提交的检查点 ID”），并帮助选举/锁定哪个 JobManager 是 leader（避免脑裂）。
配置域包括：
- 通用高可用选项（high-availability.type 等）；
- 高可用场景下 JobResultStore 的选项；
- ZooKeeper 高可用选项。
Memory Configuration（内存配置）
控制 TaskManager / JobManager 的内存使用方式。Flink 尽量屏蔽 JVM 配置复杂度。多数情况下，用户只需设置 taskmanager.memory.process.size 或 taskmanager.memory.flink.size（视部署方式而定），必要时用 taskmanager.memory.managed.fraction 调整 JVM Heap 与 Managed Memory 的比例；其余选项用于性能调优与排查内存错误。详见 TaskManager 内存配置与 JobManager 内存配置。
Miscellaneous Options
其余未归类的通用选项。
Security（安全）
配置 Flink 安全及与外部系统安全交互的选项。
SSL
Flink 网络连接可通过 SSL 加密，详细指南见 SSL 配置文档。
Auth with External Systems
- Delegation token：Flink 提供与认证协议无关的可插拔委派令牌框架，详见 Flink and Delegation Token 文档。
- ZooKeeper Authentication / Authorization：连接启用安全认证的 ZooKeeper quorum 时必需的选项。
- Kerberos-based Authentication / Authorization：设置指南及可通过 Kerberos 认证的外部系统列表见 Flink and Kerberos 文档。
Resource Orchestration Frameworks（资源编排框架）
与 Kubernetes、Yarn 等资源编排框架集成的选项。注意并非总是需要集成：例如完全可以不告知 Flink 它运行在 Kubernetes 上（不指定任何 K8s 配置项）就完成部署，参见 standalone Kubernetes 指南。本节选项仅在 Flink 主动向编排器申请和释放资源时必需。
- YARN：yarn.* 前缀选项。
- Kubernetes：kubernetes.* 前缀选项。
State Backends（状态后端）
背景知识见 State Backend 文档。
RocksDB State Backend
配置 RocksDB 状态后端最常用的选项。高级底层配置与故障排查选项见下文 “Advanced RocksDB State Backends Options”。
RocksDB 原生指标可细分为两类（用于 RocksDB 状态后端的作业时生效）：
⚠️ 启用 RocksDB 原生指标可能导致性能下降，请谨慎开启。
Metrics（指标）
背景知识见 metrics 系统文档。Flink 的指标基础设施通过 metrics.* 前缀配置，支持接入 Prometheus、InfluxDB、Graphite、JMX 等各类 Reporter。
Traces（追踪）
背景知识见 tracing 系统文档。Flink 通过 traces.* 前缀配置追踪上报。
History Server
History Server 保存已完成作业的信息（执行图、运行时长、统计信息）。启用它需先在 JobManager 开启“作业归档”（jobmanager.archive.fs.dir），详见 History Server 文档。
Experimental（实验性）
实验性功能选项。
Client
客户端相关配置，如连接超时、重试等。
User Artifact Management（用户制品管理）
Flink 在 Application 模式下可上传和拉取本地用户制品（如作业归档、单独打包的 UDF）：
💡 制品拉取（Artifact Fetching）支持 Standalone Application 模式与 Native Kubernetes Application 模式。
Execution（执行）
包含部署与执行相关配置：
- Pipeline：pipeline.* 前缀，如 pipeline.name、pipeline.auto-watermark-interval；
- Checkpointing：execution.checkpointing.*，见 CheckpointingOptions 定义；
- Recovery：execution.state-recovery.*。
如前所述，Session 集群上通过配置文件只能调整这部分 Execution 参数（影响作业），而非底层集群本身。
Debugging & Expert Tuning（调试与专家调优）
⚠️ 以下选项面向专家用户，用于排查/修复问题。大多数部署无需配置。
- Class Loading：Flink 动态加载提交到 Session 集群的作业代码，并尽量从类路径中隐藏许多依赖以减少依赖冲突，详见 Debugging Classloading 文档；
- Advanced Options for debugging：调试与调优的高级选项；
- Advanced Checkpointing Options：execution.checkpointing.* 高级选项（如 unaligned、tolerable-failed-checkpoints）；
- State Latency Tracking Options：状态访问延迟追踪；
- Advanced RocksDB State Backends Options：RocksDB 及其检查点的高级调优；
- State Changelog Options：状态变更日志（Changelog），见 State Backends 的 Changelog 章节；当 state.changelog.storage 设为 filesystem 时，使用 FileSystem-based Changelog 选项；
- RocksDB Configurable Options：对 ColumnFamily 行为与资源的细粒度控制。自 Flink 1.10 引入 state.backend.rocksdb.memory.managed 与 state.backend.rocksdb.memory.fixed-per-slot 后，这些选项通常仅在高级性能调优时使用，也可在程序中通过 RocksDBStateBackend.setRocksDBOptions(RocksDBOptionsFactory) 指定；
- Advanced Fault Tolerance Options：与故障切换、组件误判相关的问题；
- Advanced Cluster / JobManager / Scheduling / HA / HA-ZooKeeper / HA-Kubernetes / SSL Options：集群、JobManager、调度、高可用、SSL 的进阶选项；
- Advanced Options for REST endpoint and Client / Flink Web UI：REST 端点、客户端与 Web UI 的高级选项；
- Full JobManager Options：包含 Blob Server（JobManager 内用于分发过大无法附加到 RPC 消息、且受益于缓存的对象，如 jar 文件或大型序列化代码对象）与 ResourceManager（与资源编排框架无关的基础行为）；
- Full TaskManagerOptions：完整 TaskManager 选项，其中 taskmanager.network.memory.buffer-debloat.* 用法见 网络内存调优指南，另有处理 TaskManager 间流/批数据交换的 Data Transport Network Stack 选项；
- RPC / Pekko：Flink 组件间（JobManager/TaskManager/ResourceManager）使用 Pekko 做 RPC，但 Pekko 不用于数据传输。
JVM and Logging Options
env.* 前缀的环境与日志选项，例如 env.java.opts.all、env.log.dir 等。仓库默认的 config.yaml 模板 中即包含为 Java 17 支持的 env.java.opts.all 设置（Java 8/11 下可安全移除）。
Forwarding Environment Variables（转发环境变量）
可配置在 Yarn 上启动的 JobManager / TaskManager 进程的环境变量：
- containerized.master.env.：给 JobManager 进程传递自定义环境变量的前缀。例如给 JobManager 传递 LD_LIBRARY_PATH：
containerized.master.env.LD_LIBRARY_PATH: "/usr/lib/native"
- containerized.taskmanager.env.：类似前缀，为 worker（TaskManager）设置自定义环境变量。
Deprecated Options（已废弃选项）
与不再积极开发的功能相关的选项，可能在未来版本移除：DataSet API Optimizer、DataSet API Runtime Algorithms、DataSet File Sinks 等 DataSet API 相关配置。
实战小结：一份可直接落地的 config.yaml
结合仓库自带的 config.yaml 模板 与上文各配置域，一份典型的生产配置可组织为：
# Java 环境（Java 17 需要；Java 8/11 可移除）
env:
java:
opts:
all: –add-opens=java.base/java.lang=ALL-UNNAMED
# 基础：JobManager / TaskManager
jobmanager:
rpc:
address: jobmanager-host
port: 6123
memory:
process:
size: 1600m
taskmanager:
numberOfTaskSlots: 4
memory:
process:
size: 1728m
parallelism:
default: 4
# REST / Web UI
rest:
address: 0.0.0.0
port: 8081
# 高可用（以 ZooKeeper 为例）
high-availability:
type: zookeeper
zookeeper:
quorum: zk1:2181,zk2:2181,zk3:2181
storageDir: hdfs:///flink/ha/
# 容错与检查点
execution:
checkpointing:
interval: 3min
mode: EXACTLY_ONCE
timeout: 10min
state:
backend:
type: rocksdb
checkpoints:
dir: hdfs://namenode-host:port/flink-checkpoints
savepoints:
dir: hdfs://namenode-host:port/flink-savepoints
restart-strategy:
type: failure-rate
failure-rate:
delay: 1 s
failure-rate-interval: 1 min
max-failures-per-interval: 3
配置生效流程：配置文件在 Flink 进程启动时被解析（核心解析器位于 flink-core 下的 Configuration、ConfigUtils 与各类 *Options 定义类中），所有配置项最终对应一个带默认值的 ConfigOption；修改配置文件后需重启相关进程。若发现配置未按预期生效，请优先检查：是否使用了正确的配置文件名、是否存在重复 Key（新格式会直接报错）、value 类型是否被引号包裹成了字符串等。
赞
【免费下载链接】flink
项目地址：
https://gitcode.com/gh_mirrors/fli/flink
点击查看 免费下载
相关推荐
纪念币预约神器：告别手速不够的烦恼，轻松搞定限量收藏
GPT-SoVITS预训练模型完全指南：3步搞定6个版本的下载、配置与更新
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
