文章目录一. 概述二. Flink配置MinIO实现Checkpoint和Savepoint1. 配置s3文件系统2. 配置checkpoint和savepoint3. 提交一个flink job到flink集群上三. minio的安装1.单节点安装2.多节点安装多节点minio安装 ing配置tegine三. flink的高可用安装1. 配置flink-conf.yaml2. Masters 和workers设置3. 文件分发并修改4. 启动一. 概述先看下flink 的几种部署模式和适用场景本文通过使用flinkMinIO安装实现flink standalone的集群模式实现“轻量化集群”flink集群本身作为计算资源去执行flink jobMinio 用于存储Flink job产生checkpoint和savepoint、以及存储flink HA的一些信息也就是作为分布式存储系统。二. Flink配置MinIO实现Checkpoint和Savepoint1. 配置s3文件系统Apache Flink 1.12 Documentation: Amazon S3[Apache Flink]: Where is flink-s3-fs-hadoop plugin?将opt的文件复制到plugins下如下图2. 配置checkpoint和savepointflink-conf.yaml配置文件修改fs.allowed-fallback-filesystems:s3state.backend:filesystemstate.checkpoints.dir:s3://state/checkpointstate.savepoints.dir:s3://state/savepoints3.endpoint:http://10.0.16.220:9000s3.path.style.access:trues3.access-key:minios3.secret-key:minio123配置checkpoint目录时minio上要提前创建好bucket并且不能使用bucket作为checkpoint根路径。否则不能实例化jobmanager。本文是state。3. 提交一个flink job到flink集群上三. minio的安装1.单节点安装#新建minio文件夹mkdirminiocdminio#新建存数据的文件夹mkdirdata#下载minio 如果链接失效访问mino官网获取最新下载地址wgethttps://dl.min.io/server/minio/release/linux-amd64/minio#加权限chmodx minio#自定义账号密码exportMINIO_ACCESS_KEYminioexportMINIO_SECRET_KEYminio123后台启动nohup/data/a 2pp/minio/minio server /data/app/minio/data/data/log/minio/minio.log21前端访问ip:9000mac下安装2.多节点安装多节点minio安装 ing配置tegine为了不将所有的读写请求都落到一个节点造成读写压力通过利用tengine的反向代理实现负载均衡#配置minio的upstream主要实现集群的反向代理和请求转发cd/data/app/tengine/conf/vhostvimhttp_minio.proxy.conf#重新启动teginebash/data/app/tengine/scripts/tengine restart# 配置http_minio.proxy.conf# Minio 主备 配置upstream minio{server10.0.16.214:18125weight1;server10.0.16.231:28125weight1;server10.0.16.220:18125weight1;server10.0.16.247:18125weight1;checkinterval1000rise2fall5timeout1000typehttp;check_http_sendGET /minio/health/live HTTP/1.0\r\n\r\n;check_http_expect_alive http_2xx http_3xx;}# Minio Server 域server{# tegine 监听minio请求的端口和tegine的iplisten18095;server_name10.0.16.214;ignore_invalid_headers off;client_max_body_size0;proxy_buffering off;location /{proxy_set_header X-Real-IP$remote_addr;proxy_set_header X-Forwarded-For$proxy_add_x_forwarded_for;proxy_set_header X-Forwarded-Proto$scheme;proxy_set_header Host$http_host;proxy_http_version1.1;proxy_set_header Connection;proxy_connect_timeout300;proxy_pass http://minio/;}}三. flink的高可用安装设置三个节点的flink实现高可用。集群角色分布如下flink-ip1jobmanager taskmanagerflink-ip2jobmanager taskmanagerflink-ip3jobmanager taskmanager1. 配置flink-conf.yaml## 1. Common 配置## The external address of the host on which the JobManager runs and can be# reached by the TaskManagers and any clients which want to connect. This setting# is only used in Standalone mode and may be overwritten on the JobManager side# by specifying the --host hostname parameter of the bin/jobmanager.sh executable.# In high availability mode, if you use the bin/start-cluster.sh script and setup# the conf/masters file, this will be taken care of automatically. Yarn/Mesos# automatically configure the host name based on the hostname of the node where the# JobManager runs.# 指定所在节点的ipjobmanager.rpc.address:10.2.3.155# The RPC port where the JobManager is reachable.jobmanager.rpc.port:18550jobmanager.rpc.bind-port:18550# The total process memory size for the JobManager.## Note this accounts for all memory usage within the JobManager process, including JVM metaspace and other overhead.jobmanager.memory.process.size:1600m# The total process memory size for the TaskManager.## Note this accounts for all memory usage within the TaskManager process, including JVM metaspace and other overhead.taskmanager.memory.process.size:1728m# To exclude JVM metaspace and overhead, please, use total Flink memory size instead of taskmanager.memory.process.size.# It is not recommended to set both taskmanager.memory.process.size and Flink memory.## taskmanager.memory.flink.size: 1280m# The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.taskmanager.numberOfTaskSlots:8# The parallelism used for programs that did not specify and other parallelism.parallelism.default:1## 2. High Availability# 以下配置在standalone模式下打开## ACL options are based on https://zookeeper.apache.org/doc/r3.1.2/zookeeperProgrammers.html#sc_BuiltinACLSchemes# It can be either creator (ZOO_CREATE_ALL_ACL) or open (ZOO_OPEN_ACL_UNSAFE)# The default value is open and it can be changed to creator if ZK security is enabled## high-availability.zookeeper.client.acl: openhigh-availability:zookeeperhigh-availability.storageDir:s3://flink/ha/high-availability.zookeeper.quorum:10.2.3.59:127,10.2.3.156:18127,10.2.3.155:18127high-availability.zookeeper.client.acl:open### 3. 设置checkpoint 和savepointfs.allowed-fallback-filesystems:s3state.backend:filesystemstate.checkpoints.dir:s3://state/checkpointstate.savepoints.dir:s3://state/savepoints3.endpoint:http://xxx:18095/s3.path.style.access:trues3.access-key:admins3.secret-key:MINIOminio_123## 4. Fault tolerance and checkpointing## The backend that will be used to store operator state checkpoints if# checkpointing is enabled.jobmanager.execution.failover-strategy:region## Rest web frontend## The port to which the REST client connects to. If rest.bind-port has# not been specified, then the server will bind to this port as well.#rest.port:18551# The address to which the REST client will connect to##rest.address: 0.0.0.0# Port range for the REST and web server to bind to.#rest.bind-port:18551-18590# The address that the REST web server binds to##rest.bind-address: 0.0.0.0# Flag to specify whether job submission is enabled from the web-based# runtime monitor. Uncomment to disable.web.submit.enable:true## Advanced##flink的ssh参数env.ssh.opts:-p 36000##flink日志路径env.log.dir:/data/logs/flink## 监控设置metrics.reporter.promgateway.password:Push$777metrics.reporter.promgateway.metric.only.report.names:flink_jobmanager_job_lastCheckpointSize,flink_jobmanager_job_lastCheckpointDuration,flink_jobmanager_job_numberOfFailedCheckpoints,flink_jobmanager_Status_JVM_CPU_Load,flink_jobmanager_Status_JVM_Memory_Heap_Used,flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Count,flink_jobmanager_Status_JVM_GarbageCollector_MarkSweepCompact_Count,flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Time,flink_jobmanager_Status_JVM_GarbageCollector_MarkSweepCompact_Time,flink_taskmanager_Status_JVM_CPU_Load,flink_taskmanager_Status_JVM_Memory_Heap_Used,flink_taskmanager_Status_JVM_GarbageCollector_MarkSweepCompact_Count,flink_taskmanager_Status_JVM_GarbageCollector_G1_Old_Generation_Count,flink_taskmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Count,flink_taskmanager_Status_JVM_GarbageCollector_MarkSweepCompact_Time,flink_taskmanager_Status_JVM_GarbageCollector_G1_Old_Generation_Time,flink_taskmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Time,flink_taskmanager_job_task_Shuffle_Netty_Output_Buffers_outPoolUsage,flink_taskmanager_job_task_Shuffle_Netty_Input_Buffers_inPoolUsage,flink_taskmanager_job_task_Shuffle_Netty_Output_Buffers_outputQueueLength,flink_taskmanager_job_task_Shuffle_Netty_Input_Buffers_inputQueueLength,flink_taskmanager_job_task_currentInputWatermark,flink_jobmanager_job_downtime,flink_taskmanager_job_task_numBytesIn,flink_taskmanager_job_task_numBytesOut,flink_taskmanager_job_task_numRecordsIn,flink_taskmanager_job_task_numRecordsOutinternal.yarn.log-config-file:/data/app/flink/conf/log4j.propertiesmetrics.reporter.promgateway.host:promgateway-ipmetrics.reporter.promgateway.username:adminmetrics.reporter.promgateway.deleteOnShutdown:falsemetrics.reporter.promgateway.class:org.apache.flink.metrics.prometheus.CustomedPrometheusPushGatewayReportermetrics.reporter.promgateway.groupingKey:mode2;id100000153;account_id110;instancemetrics.reporter.promgateway.jobName:myJobmetrics.reporter.promgateway.port:18165metrics.reporter.promgateway.interval:15 SECONDS注意省略zookeeper的安装2. Masters 和workers设置masters代表三个节点都有jobmanager角色对于高可用来说其中两个jobmanager挂掉flink集群仍然可以运行。[root10-101-1-48 conf]# cat mastersflink-ip1:18551 flink-ip2:18551 flink-ip3:18551[root10-101-1-48 conf]# cat workersflink-ip1 flink-ip2 flink-ip33. 文件分发并修改scp-P36000/data/app/flink/conf/masters userflink-ip:/data/app/flink/confscp-P36000/data/app/flink/conf/workers userflink-ip:/data/app/flink/confscp-P36000/data/app/flink/conf/flink-conf.yaml userflink-ip:/data/app/flink/conf修改flink-conf.yaml下的# 指定所在节点的ipjobmanager.rpc.address:ip4. 启动三个节点分别执行bash /data/app/flink/bin/jobmanager.sh start bash /data/app/flink/bin/taskmanager.sh start