Flink 配置文件说明

r囧r小猫 2021-09-23 15:02 583阅读 0赞

## master ##

集群的 master 节点

## slaves ##

集群中所有的工作节点

## flink-conf.yaml ##

### 1. 基础配置 ###

# JobMaster节点IP地址
    jobmanager.rpc.address
    
    # JobMaster端口号，默认6123
    jobmanager.rpc.port：6123
    
    # JM 节点可分配的最大内存，master 主要管理集群资源(RM)、协调应用程序执行(TM)，需要资源不多，
    # 但如果多个应用并行或者一个应用有很多算子，那么需要增加内存，默认1024m
    jobmanager.heap.size: 1024m
    
    # Worker 节点负责实际任务的执行并处理大量数据，而且当使用内存或文件格式的状态后端，也会占用 JVM 堆内存
    # 默认1024m，根据实际应用情况适当调整。On YARN 模式下，内存大小是被 TM 的 YARN 资源容器自动分配的
    taskmanager.heap.size: 1024m
    
    # 每个 TM 提供的任务 Slots 数量大小，即 Work 进程可以同时运行的 Task 数，通常是 CPU 的核心数或者一半
    # 默认1。Flink 不为每个 Task 或 Solt 指定分配内存，所以一个任务也可能消耗 Work JVM 大部分内存
    taskmanager.numberOfTaskSlots: 1
    
    # Flink 默认并行计算的个数，默认1。即应用内每个算子的并行度，与整个集群的 CPU 核心数有关，
    # 整个集群最大可以的并行度, 一般设置为 slave节点数 * 节点可用CPU核数
    # 增加 parallelism 可以提高任务并行的计算的实例数，提升数据处理效率，但也会占用更多 Slot
    # 算子的并行度有多种配置方式，优先级是：
    # 配置文件(parallelism.default) < 启动时-p参数指定 < 代码中初始设置(env.setParallelism) < 代码中算子设置(setParallelism)
    parallelism.default: 1
    
    # 文件系统来源
    fs.default-scheme

### 2. 高可用性配置 ###

# 可选择 'NONE' 或者 'zookeeper'
    high-availability: zookeeper
    
    # Flink 在高可用时持久化元数据位置
    high-availability.storageDir: hdfs:///flink/ha/
    
    # zookeeper 集群节点 ip 和 port，逗号分割
    high-availability.zookeeper.quorum: localhost:2181
    
    # 默认 open，如果 zookeeper security 启用了该值会更改成 creator
    high-availability.zookeeper.client.acl: open

### 3. 容错和检查点配置 ###

# 用于存储和检查点状态，会影响应用程序的性能
    # jobmanager（MemoryStateBackend）：存储于JM内存，效率最高但不安全，但一般测试使用，
    # 如果丢失，一旦任务失败无法通过 checkpoint 重启，
    # 可选模式：
    # filesystem（FsStateBackend）：存储于文件系统
    # rocksdb（RocksDBStateBackend）：存储于数据库
    state.backend: filesystem
    
    # 自动存储检查点的数据文件和元数据的默认目录，这个目录必须能被所有的进程、节点访问
    state.checkpoints.dir: hdfs://namenode-host:port/flink-checkpoints
    
    # 手动保存状态时 savepoints 的目录(可选)，用于状态存储写文件时使用（MemoryStateBackend，FsStateBackend，
    # RocksDBStateBackend）
    state.savepoints.dir: hdfs://namenode-host:port/flink-checkpoints
    
    # 是否启用增量 checkpoints（如rocksdb模式），增量 checkpoint 仅存储和之前checkpoint的差异，
    # 而不是完整的checkpoint。但不是所有的 backend 状态都支持异步和增量 checkpoint
    state.backend.incremental: false
    
    # Task故障策略，Flink1.9合并Blink后功能，Flink 将中间结果保留在网络 shuffle 的边缘，
    # 并使用此数据去恢复那些仅受故障影响的 Task，缩短了 Task 失败后恢复的时间
    jobmanager.execution.failover-strategy: region

### 4. Rest和网络配置 ###

# Web 的运行时监视器端口
    rest.port: 8081
    
    # Web 的运行时监视器侦听的地址
    jobmanager.web.address: 0.0.0.0
    
    # 是否从基于 Web 的 jobmanager 启用作业提交
    web.submit.enable: false

### 5. 高级配置 ###

# 指定默认的本地临时目录。Flink在运行时会把临时数据写到本地文件系统，
    # 比如 Flink 接收到的JAR、应用程序状态(当用RocksDB存储应用程序状态时)，要避免目录里的数据被服务器自动清空
    # 否则 job 重启时可能因找不到元数据导致恢复失败
    io.tmp.dirs: /tmp 
    
    # 是否在 TM 启动时预先分配 TM 管理的内存
    taskmanager.memory.preallocate: false
    
    # 类加载解析顺序，先检查用户代码 jar（“child-first”）还是应用程序类路径（“parent-first”）
    # 默认设置指示首先从用户代码 jar 加载类
    classloader.resolve-order: child-first
    
    # TM 节点为网络缓冲区分配的内存，默认占JVM堆内存10%，
    # 这决定了 TM 可以同时拥有多少流数据交换通道以及通道缓冲的程度
    # 如果作业被拒绝或者收到系统没有足够缓冲区的警告，需增加此值或最小/最大值
    # 另注意，“taskmanager.network.memory.min”和“taskmanager.network.memory.max”可能会覆盖此分数
    # TM 的网络缓存区用于跨节点的数据交换(Task接收、发送事件)，是网络交换的关键组件。
    # TM 在发送事件之前先把事件收集到缓存区、在接收到数据之后并把数据传递给应用之前缓存到缓冲区。
    # 这种方式有效的利用网络资源实现高吞吐。缓存区的总数量和任务之间的网络连接总数相同，比如通过分区或广播连接起来的两个算子，
    # 缓存区总数是发送和接收算子的并行度的乘积。网络缓存区在JVM堆外内存分配
    # 如果没有足够数量的缓存区，可能不能为同时建立的所有网络连接分配缓存区，会降低吞吐量
    taskmanager.network.memory.fraction: 0.1
    
    # 网络缓冲区最小值
    taskmanager.network.memory.min: 64mb
    
    # 网络缓冲区最大值
    taskmanager.network.memory.max: 1gb
    
    # 配置一个网络堆栈分配的内存大小，默认32k
    # 减少网络堆栈(一个网络数据交换管道的网络缓冲区)会增加网络缓冲区数量，但有可能降低整体的网络传输效率
    taskmanager.memory.segment-size：32k
    
    # TM 内存配置
    taskmanager.memory.size：
    
    # 按照比例分配内存。TM 为排序、HashTable、中间结果的缓存而分配的内存。
    taskmanager.memory.fraction：
    
    # TM 堆外内存
    taskmanager.memory.off-heap：

### 6. Flink 集群安全配置 ###

# 是否从 Kerberos ticket 缓存中读取
    security.kerberos.login.use-ticket-cache: true
    
    # 包含用户凭据的 Kerberos 密钥表文件的绝对路径
    security.kerberos.login.keytab: /path/to/kerberos/keytab
    
    # 与 keytab 关联的 Kerberos 主体名称
    security.kerberos.login.principal: flink-user
    
    # 以逗号分隔的登录上下文列表，用于提供 Kerberos 凭据
    # 例如，Client，KafkaClient 使用凭证进行 ZooKeeper 身份验证和 Kafka 身份验证
    security.kerberos.login.contexts: Client,KafkaClient

### 7. Zookeeper 安全配置 ###

# 覆盖以下配置以提供自定义 ZK 服务名称
    zookeeper.sasl.service-name: zookeeper
    
    # 该配置必须匹配 security.kerberos.login.contexts 配置中的列表（含有一个）
    zookeeper.sasl.login-context-name: Client

### 8. HistoryServer ###

# 可以通过 bin/historyserver.sh (start|stop) 命令启动和关闭 HistoryServer服务
    # 已完成作业上传到目录
    jobmanager.archive.fs.dir: hdfs:///completed-jobs/
    
    # 基于 Web 的 HistoryServer 的地址
    historyserver.web.address: 0.0.0.0
    
    # 基于 Web 的 HistoryServer 的端口号
    historyserver.web.port: 8082
    
    # 以逗号分隔的目录列表，用于监视已完成的作业
    historyserver.archive.fs.dir: hdfs:///completed-jobs/
    
    # 刷新受监控目录的时间间隔（毫秒）
    historyserver.archive.fs.refresh-interval: 10000