flink集群(docker版)配置及使用
1 环境说明
注意：以下所有操作都在root用户下完成 sudo su – root
|
ip |
操作系统版本 |
用途 |
|
192.168.30.18 |
Ubuntu 18.04.4 LTS |
jobmanager容器、nfs服务(存储flink的checkpoint、savepoint) |
|
192.168.30.17 |
Ubuntu 18.04.4 LTS |
taskmanager02容器 |
|
192.168.30.16 |
Ubuntu 18.04.4 LTS |
taskmanager01容器 |
2 部署nfs
在192.168.30.18节点上操作
2.1安装nfs软件包
# apt-get install rpcbind -y
# apt-get install nfs-kernel-server -y
2.2 配置参数，映射checkpoints存储目录
# vim /etc/exports
/home/liuchan/config/flink/ 192.168.30.0/24(rw,sync,no_root_squash,no_all_squash)
2.3 重启nfs服务并加入开机自启动
# systemctl restart rpcbind
# systemctl enable rpcbind
# systemctl restart nfs-kernel-server
# systemctl enable nfs-kernel-server
2.4 查看nfs状态
# rpcinfo -p localhost
program vers proto port service
100000 4 tcp 111 portmapper
100000 3 tcp 111 portmapper
100000 2 tcp 111 portmapper
100000 4 udp 111 portmapper
100000 3 udp 111 portmapper
100000 2 udp 111 portmapper
100005 1 udp 34715 mountd
100005 1 tcp 46897 mountd
100005 2 udp 48806 mountd
100005 2 tcp 39469 mountd
100005 3 udp 41227 mountd
100005 3 tcp 34733 mountd
100003 3 tcp 2049 nfs
100003 4 tcp 2049 nfs
100227 3 tcp 2049
100003 3 udp 2049 nfs
100227 3 udp 2049
100021 1 udp 35642 nlockmgr
100021 3 udp 35642 nlockmgr
100021 4 udp 35642 nlockmgr
100021 1 tcp 42347 nlockmgr
100021 3 tcp 42347 nlockmgr
100021 4 tcp 42347 nlockmgr
2.5 查看nfs映射信息
# exportfs -v
/home/liuchan/config/flink
192.168.30.0/24(rw,wdelay,no_root_squash,no_subtree_check,sec=sys,rw,secure,no_root_squash,no_all_squash)
3 客户端挂载nfs
在192.168.30.16、17节点上操作
3.1 安装nfs挂载工具
# apt install nfs-common
# showmount -e 192.168.30.18
Export list for 192.168.30.18:
/home/liuchan/config/flink 192.168.30.0/24
3.2 创建挂载目录
# mkdir -p /home/liuchan/config/flink
3.3 挂载nfs并写入到/etc/fstab
# echo ‘192.168.30.18:/home/liuchan/config/flink /home/liuchan/config/flink nfs defaults 0 0’ >> /etc/fstab
# mount -a
4 集群通用配置说明
4.1 docker-compose说明
1 compose目录结构
# tree -LFa 1 flink/
flink
├── conf/ # flink容器配置文件目录
├── docker-compose.yml # docker-compose配置文件
├── Dockerfile # 构建flink镜像用到的Dockerfile文件
├── .env # docker-compose环境变量文件
├── flink-1.10.0.tar.gz # 构建flink镜像用到的flink-1.10.0压缩包
├── jdk1.8.0_251.tar.gz # 构建flink镜像用到的jdk1.8.0_251压缩包
├── log/ # flink容器日志目录
├── run.sh # 构建flink镜像用到的flink容器启动脚本
└── tmp/ # flink容器临时文件目录
3 directories, 6 files
2 .env文件
# cat flink/.env
IMAGE=flink:1.10.0-v01
FLINK_HOME=/usr/local/flink-1.10.0
3 Dockerfile文件
# cat flink/Dockerfile
FROM centos:7
LABEL maintainer lc
ARG FLINK_VERSION=flink-1.10.0
ARG JDK_VERSION=jdk1.8.0_251
RUN yum install wget curl unzip iproute net-tools -y && \
yum clean all && \
rm -rf /var/cache/yum/* && \
ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
ADD ${FLINK_VERSION}.tar.gz /usr/local
ADD ${JDK_VERSION}.tar.gz /usr/local
ENV JAVA_HOME /usr/local/${JDK_VERSION}
ENV CLASSPATH $JAVA_HOME/lib/tools.jar:$JAVA_HOME/jre/lib/rt.jar
ENV PATH $JAVA_HOME/bin:$PATH
ENV FLINK_HOME=/usr/local/${FLINK_VERSION}
COPY run.sh /root
WORKDIR ${FLINK_HOME}
EXPOSE 8081 6123
ENTRYPOINT [“/root/run.sh”]
4 run.sh文件
#!/bin/bash
FLINK_JOB_MANAGER_SH=$FLINK_HOME/bin/jobmanager.sh
FLINK_TASK_MANAGER_SH=$FLINK_HOME/bin/taskmanager.sh
case “$1” in
“jobmanager”)
$FLINK_JOB_MANAGER_SH start-foreground
;;
“taskmanager”)
$FLINK_TASK_MANAGER_SH start-foreground
;;
*)
echo “COMMAND ERROR”
;;
esac
4.2 flink集群配置文件
1 修改flink-conf.yaml
# vim flink/conf/flink-conf.yaml
# jobmanager节点地址
jobmanager.rpc.address: 192.168.30.18
# checkpoints保留的个数，全局配置
state.checkpoints.num-retained: 16
# Blog Server
taskmanager.rpc.port: 40010-40020
# Task Manager
metrics.internal.query-service.port: 40030-40040
# Metrics
blob.server.port: 40050-40060
2 修改masters
# vim flink/conf/masters
192.168.30.18:8081
3 修改slaves
# cat flink/conf/slaves
192.168.30.16
192.168.30.17
5 启动容器
5.1 jobmanager
192.168.30.18节点
1 docker-compose.yml配置文件
# cat flink/docker-compose.yml
version: ‘3’
networks:
rulr-network:
external: true
services:
flink-1.10.0-v01:
image: flink:1.10.0-v01
build:
context: ./
dockerfile: Dockerfile
jobmanager:
container_name: jobmanager
hostname: jobmanager
image: ${IMAGE}
volumes:
– ./conf:${FLINK_HOME}/conf
– ./tmp/jobmanager-tmp:/tmp
– ./log/jobmanager-log:${FLINK_HOME}/log
– /home/liuchan/config/flink:/home/liuchan/config/flink
– /home/liuchan/servers/compute-streaming:/home/liuchan/servers/compute-streaming
– /home/liuchan/config/servers:/home/liuchan/config/servers
ports:
– “8081:8081”
– “6123:6123”
– “40010-40020:40010-40020”
– “40030-40040:40030-40040”
– “40050-40060:40050-40060”
command: jobmanager
restart: always
networks:
– rulr-network
deploy:
resources:
limits:
cpus: ‘2’
memory: 2G
2 构建容器
# cd flink-jobmanager/
# docker-compose build
# docker-compose up -d
5.2 taskmanager02
192.168.30.17节点
1 docker-compose.yml配置文件
# cat flink/docker-compose.yml
version: ‘3’
networks:
rulr-network:
external: true
services:
flink-1.10.0-v01:
image: flink:1.10.0-v01
build:
context: ./
dockerfile: Dockerfile
taskmanager02:
container_name: taskmanager02
hostname: taskmanager02
image: ${IMAGE}
ports:
– “40010-40020:40010-40020”
– “40030-40040:40030-40040”
– “40050-40060:40050-40060”
volumes:
– ./conf:${FLINK_HOME}/conf
– ./tmp/taskmanager02-tmp:/tmp
– ./log/taskmanager02-log:${FLINK_HOME}/log
– /home/liuchan/config/flink:/home/liuchan/config/flink
command: taskmanager
restart: always
networks:
– rulr-network
deploy:
resources:
limits:
cpus: ‘2’
memory: 2G
2 构建容器
# cd flink-taskmanager02/
# docker-compose build
# docker-compose up -d
5.3 taskmanager01
192.168.30.16节点
1 docker-compose.yml配置文件
# cat flink/docker-compose.yml
version: ‘3’
networks:
rulr-network:
external: true
services:
flink-1.10.0-v01:
image: flink:1.10.0-v01
build:
context: ./
dockerfile: Dockerfile
taskmanager01:
container_name: taskmanager01
hostname: taskmanager01
image: ${IMAGE}
ports:
– “40010-40020:40010-40020”
– “40030-40040:40030-40040”
– “40050-40060:40050-40060”
volumes:
– ./conf:${FLINK_HOME}/conf
– ./tmp/taskmanager01-tmp:/tmp
– ./log/taskmanager01-log:${FLINK_HOME}/log
– /home/liuchan/config/flink:/home/liuchan/config/flink
command: taskmanager
restart: always
networks:
– rulr-network
deploy:
resources:
limits:
cpus: ‘2’
memory: 2G
2 构建容器
# cd flink-taskmanager01/
# docker-compose build
# docker-compose up -d
6 运行flink任务
6.1 运行jar包
# docker exec -it flink-jobmanager bash
[root@jobmanager flink-1.10.0]# bin/flink run -d /home/liuchan/servers/compute-streaming/compute-streaming.jar –appMode=2
6.2 在flink的UI页面查看
URL: http://192.168.30.18:8081/
# 总体概览
# Task Managers信息
# 查看JOB详细信息
6.3 测试
1 停掉正在运行job的TaskManager容器
从上面的截图可以判断出ip为172.18.0.10容器正在运行job，该容器在192.168.30.17服务器上。停掉taskmanager02容器。
# cd flink-taskmanager02/
# docker-compose down
2 在flink的UI中查看
从下图可以看到，job经历超时报错后，运行的TaskManager节点由taskmanager02容器转移到了taskmanager01容器
3 恢复taskmanager02容器
# cd flink-taskmanager02/
# docker-compose up -d
# 在flink的UI可以看到taskmanager02容器又上线了
7 Checkpoint、Savepoint
7.1 Checkpoint、Savepoint说明
1 Checkpoint
(1) Flink Checkpoint是一种容错恢复机制。这种机制保证了实时程序运行时，即使突然遇到异常也能够进行自我恢复，不需要用户指定。Checkpoint是增量做的，每次的时间较短，数据量较小。
(2) Checkpoint对于用户层面，是透明的，用户会感觉程序一直在运行。
(3) Flink Checkpoint是Flink自身的系统行为，用户无法对其进行交互，用户可以在程序启动之前，设置好实时程序Checkpoint相关参数，当程序启动之后，剩下的就全交给Flink自行管理。
(4) Checkpoint默认关闭，Checkpoint的启用、存储方式、存储位置，在应用代码中配置，其中存储方式、存储位置，也可以在flink-conf.yaml文件中通过state.backend、state.checkpoints.dir参数配置全局参数，但应用代码中配置优先级更高。
(5) Checkpoint 默认保留数为1，通过修改flink-conf.yaml文件中state.checkpoints.num-retained参数设置Checkpoint保留数量。
(6) Checkpoint默认job程序取消后删除，在应用代码中通过设置以下代码实现保留：
“getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);”
(5) Checkpoint的组成：Checkpoint根目录/CheckpointID
2 Savepoint
(1) Flink Savepoint你可以把它当做在某个时间点程序状态全局镜像，以后程序在进行升级，或者修改bug、并行度等情况，还能从保存的状态位继续启动恢复，需要用户指定。是全量做的，每次的时间较长，数据量较大。
(2) Flink Savepoint一般存储在文件系统上面，它需要用户主动进行触发。
(3) Savepoint会一直保存，除非用户删除。
(4) flink-conf.yaml文件state.savepoints.dir参数用以配置savepoints的存储目录，默认none。
(5) Savepoint的组成：Savepoint根目录/savepoint-jobid前六位-12随机数字及字母组合
7.2 flink常用命令
1 通用命令
(1) 命令执行
bin/flink run [OPTIONS]
options:
-d: 任务提交后，断开session，会话继续保持，即表示将job放到后台运行。
示例：后台运行job
# bin/flink run -d /home/liuchan/servers/compute-streaming/compute-streaming.jar –appMode=2
(2) 查看任务列表
bin/flink list [OPTIONS]
options:
-a: 全部显示所有程序及其作业ID
-r: 仅显示正在运行的程序及其作业ID
-s: 仅显示计划的程序及其作业ID
示例：查看正在运行的job
# bin/flink list -r
2 操作Savepoint
(1) 在取消任务时保存Savepoint
bin/flink cancel [OPTIONS]
options:
-s: 触发保存点并取消作业。目标目录是可选的。如果未指定目录，则
配置的默认目录使用(state.savepoints.dir)。
官方文档已经不推荐使用，建议使用stop。
示例：
# bin/flink cancel -s /home/liuchan/config/flink/savepoints 1af3e95778850085614d175657948369
(2) 在停止任务时保存Savepoint
bin/flink stop [OPTIONS]
options:
-p: 保存点的路径，如果没有指定目录时，将使
用默认值配置(state.savepoints.dir)。
示例：
# bin/flink stop -p /home/liuchan/config/flink/savepoints 1504a27aaecba591877a68a233ee9240
(3) 触发Savepoint
bin/flink savepoint [OPTIONS] []
options:
示例：
# bin/flink savepoint 3b74f51cc4186aa4bf5bf84e7e716d0f /home/liuchan/config/flink/savepoints
(4) 从指定的savepoint运行job
bin/flink run [OPTIONS]
options:
-s: 从savepoint路径到用于还原作业的保存点
示例：
# bin/flink run -d -s /home/liuchan/config/flink/savepoints/savepoint-1af3e9-7a3891c86538 /home/liuchan/servers/compute-streaming/compute-streaming.jar –appMode=2
3 取消和停止job的区别如下
(1) cancel()调用，立即调用作业算子的cancel()方法，以尽快取消他们。如果算子在接
到cancel()调用后没有停止，flink将开始定期中断算子线程的执行，直到所有算子停止
为止。
(2) stop()是更加优雅的方式，仅适用于source实现了stoppableFunction接口的作业。
当用户请求停止作业时，作业的所有source都将被stop()方法调用，指导所有source
正常关闭时，作业才会正常结束，这种方式，使作业正常处理完所有作业。
本文来自网络，不代表协通编程立场，如若转载，请注明出处：https://www.net2asp.com/f7ac547963.html
相关推荐
-
数据结构——栈
目录 一、栈 1.1 栈的基本概念 1.2 栈的实现 二、栈的接口实现 （1）初始化栈 （2）入栈 （3）出栈 （4）获取栈顶元素 （5）获取栈中有效元素个数 （6）检测栈是否为…
-
【进阶C语言】数据的存储（万字详解）
深度剖析数据在内存中的存储 1. 数据类型介绍 1.1 类型的基本归类： 2. 整形在内存中的存储 2.1 原码、反码、补码 2.2 大小端介绍 2.3 练习 3. 浮点型在内存中…
-
【动态规划】【二分查找】【去重】730. 统计不同回文子序列
作者推荐 视频算法专题 本文涉及知识点 动态规划汇总 二分查找算法合集 LeetCode730. 统计不同回文子序列 给你一个字符串 s ，返回 s 中不同的非空回文子序列个数 。…
-
RAG框架中的Retrieve算法评估
目录 什么是 RAG 中的 Retrieve？ Retrieve 算法评估 数据集构造 Retrieve 算法评估 BM25 Embedding Search Ensemble S…
-
【数据结构】顺序表—C语言版（数据结构开篇小菜，全网最详细！小白看一遍就学会！！！）
文章目录 🍔一、前言 🍟1. 什么是数据结构 🍔二、顺序表的概念—-线性表 🍟1. 什么是线性表 &#x…
-
7-1 子集和问题–回溯法（算法设计与分析）
7-1 子集和问题 作者 陈晓梅 单位 广东外语外贸大学 设集合S={x1,x2,…,xn}是一个正整数集合，c是一个正整数，子集和问题判定是否存在S的一个子集S1，使S1中…
-
【数据结构与算法】顺序表增删查改的实现（动态版本+文件操作）附源码
目录 一.前言 二.顺序表 1.概念及结构 2.顺序表结构体的定义 3.初始化顺序表，销毁顺序表和打印 3.接口 a.尾插 SepListpushback 头插 SepList…
-
WSL2+ubuntu 18+VsCode 配置C/C++开发环境 踩坑
1. 管理员模式打开cmd，或PowerShell ，输入 wsl –install 可能出现的错误：无法解析服务器名称或地址 解决方式：科学上网 安装WSL时遇到“…
-
【算法系列篇】双指针
文章目录 前言 什么是双指针算法 1.移动零 1.1 题目要求 1.2 做题思路 1.3 Java代码实现 2.复写零 2.1 题目要求 2.2 做题思路 2.3 Java代码实现…
-
Leetcode刷题之环形链表
莫等闲，白了少年头，空悲切。 –岳飞 目录 1.环形链表 2.环形链表Ⅱ 1.环形…
