KAFKA部署 1.单点部署zookeeper1.准备zookeeper和kafka的安装包 部署kafka需要准备Java运行时环境 http://kafka.apache.org/downloads #kafka下载官方地址 kafka_2.12-2.5.1.tgz zookeeper-3.4.12.tar.gz jdk-8u191-linux-x64.tar.gz 2.创建工作目录 [root@kafka108 ~]# mkdir /zy/{data/{kafka,zookeeper},logs/{kafka,zookeeper},app} -p 3.上传安装包 [root@kafka108 ~]# cd /zy/app [root@kafka108 app]# ll -rw-r--r-- 1 root root 61709988 Jun 13 15:18 kafka_2.12-2.5.1.tgz -rw-r--r-- 1 root root 36667596 Jun 13 15:18 zookeeper-3.4.12.tar.gz -rw-r--r-- 1 root root 191753373 Jun 11 18:43 jdk-8u191-linux-x64.tar.gz 4.解压软件 [root@kafka108 app]# tar xf jdk-8u191-linux-x64.tar.gz [root@kafka108 app]# tar xf zookeeper-3.4.12.tar.gz 5.配置jdk环境变量 [root@kafka108 app]# vim /etc/profile.d/jdk.sh #!/bin/bash export JAVA_HOME=/zy/app/jdk1.8.0_191 export JRE_HOME=${JAVA_HOME}/jre export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib export PATH=$PATH:$JAVA_HOME/bin [root@kafka108 app]# source /etc/profile.d/jdk.sh #使环境变量生效 6.验证java [root@kafka108 app]# java -version java version "1.8.0_191" Java(TM) SE Runtime Environment (build 1.8.0_191-b12) Java HotSpot(TM) 64-Bit Server VM (build 25.191-b12, mixed mode) 7.创建配置文件 [root@kafka108 conf]# cp /zy/app/zookeeper-3.4.12/conf/{zoo_sample,zoo}.cfg 8.修改配置文件 [root@kafka108 ~]# vim /zy/app/zookeeper-3.4.12/conf/zoo.cfg dataDir=/zy/data/zookeeper #定义数据目录 9.定义zookeeper环境变量并启动 [root@kafka108 bin]# vim /etc/profile.d/zk.sh #!/bin/bash export ZK_HOME=/zy/app/zookeeper-3.4.12 export PATH=$PATH:$ZK_HOME/bin [root@kafka108 bin]# source /etc/profile.d/zk.sh [root@kafka108 ~]# zkServer.sh start # 启动zookeeper服务 [root@kafka108 ~]# zkServer.sh status # 查看zookeeper服务状态 [root@kafka108 ~]# zkServer.sh stop # 停止zookeeper服务 [root@kafka108 ~]# zkServer.sh restart # 重启zookeeper服务 10.确定zookeeper服务后，测试zookeeper服务是否可用 [root@kafka108 ~]# zkCli.sh ... [zk: localhost:2181(CONNECTED) 0] ls / [zookeeper] [zk: localhost:2181(CONNECTED) 1] [root@kafka108 ~]# ss -ntl|grep 2181 [root@kafka108 ~]# jps 16725 Jps 16622 QuorumPeerMain2.集群部署zookeeper配置集群免密登录及同步脚本: 1.修改主机列表 [root@kafka108 ~]# cat /etc/hosts 10.0.0.108 kafka108 10.0.0.109 kafka109 10.0.0.110 kafka110 2.kafka108节点上生成密钥对 [root@kafka108 ~]# ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa -q 3.kafka108配置所有集群节点的免密登录 [root@kafka108 ~]# for ((host_id=108;host_id<=110;host_id++));do ssh-copy-id kafka${host_id} ;done 4.链接测试 [root@kafka108 ~]# ssh kafka110 Last login: Wed Jun 26 20:55:36 2024 from 10.0.0.1 5.所有节点安装rsync数据同步工具 [root@kafka108 ~]# yum -y install rsync [root@kafka109 ~]# yum -y install rsync [root@kafka110 ~]# yum -y install rsync 6.编写同步脚本 [root@kafka108 ~]# vim /usr/local/sbin/data_rsync.sh #!/bin/bash # Auther: Zhu Yang if [ $# -ne 1 ];then echo "Usage: $0 /path/to/file(绝对路径)" exit fi # 判断文件是否存在 if [ ! -e $1 ];then echo "[ $1 ] dir or file not find!" exit fi # 获取父路径 fullpath=`dirname $1` # 获取子路径 basename=`basename $1` # 进入到父路径 cd $fullpath for ((host_id=109;host_id<=110;host_id++)) do # 使得终端输出变为绿色 tput setaf 2 echo ===== rsyncing kafka${host_id}: $basename ===== # 使得终端恢复原来的颜色 tput setaf 7 # 将数据同步到其他两个节点 rsync -apz $basename `whoami`@kafka${host_id}:$fullpath if [ $? -eq 0 ];then echo "命令执行成功!" fi done [root@kafka108 ~]# chmod +x /usr/local/sbin/data_rsync.sh #给脚本授权 7.修改配置文件 [root@kafka108 ~]# vim /zy/app/zookeeper-3.4.12/conf/zoo.cfg # 定义最小单元的时间范围tick。 tickTime=2000 # 启动时最长等待tick数量。 initLimit=10 # 数据同步时最长等待的tick时间进行响应ACK syncLimit=5 # 指定数据目录 dataDir=/zy/data/zookeeper #指定日志目录 dataLogDir=/zy/logs/zookeeper # 监听端口 clientPort=2181 # 开启四字命令允许所有的节点访问。 4lw.commands.whitelist=* # server.ID=A:B:C[:D] # ID: # zk的唯一编号。 # A: # zk的主机地址。 # B: # leader的选举端口，是谁leader角色，就会监听该端口。 # C: # 数据通信端口。 # D: # 可选配置，指定角色。 server.0=10.0.0.108:2888:3888 server.1=10.0.0.109:2888:3888 server.2=10.0.0.110:2888:3888 8.同步zookeeper软件包及目录结构 [root@kafka108 app]# data_rsync.sh /zy/app/jdk1.8.0_191/ [root@kafka108 app]# data_rsync.sh /etc/profile.d/jdk.sh [root@kafka108 app]# data_rsync.sh /etc/profile.d/zk.sh 9.去其他节点使环境变量生效 [root@kafka109 ~]# source /etc/profile.d/jdk.sh [root@kafka110 ~]# source /etc/profile.d/jdk.sh [root@kafka109 ~]# source /etc/profile.d/zk.sh [root@kafka110 ~]# source /etc/profile.d/zk.sh 10.每个zookeeper节点创建唯一标识ID [root@kafka108 ~]# vim /zy/data/zookeeper/myid 0 [root@kafka109 ~]# vim /zy/data/zookeeper/myid 1 [root@kafka110 ~]# vim /zy/data/zookeeper/myid 2 11.编写zookeeper集群管理脚本 [root@kafka108 ~]# vim /usr/local/sbin/manager_zk.sh #!/bin/bash #判断用户是否传参 if [ $# -ne 1 ];then echo "无效参数，用法为: $0 {start|stop|restart|status}" exit fi #获取用户输入的命令 cmd=$1 #定义函数功能 function zookeeperManger(){ case $cmd in start) echo "启动服务" remoteExecution start ;; stop) echo "停止服务" remoteExecution stop ;; restart) echo "重启服务" remoteExecution restart ;; status) echo "查看状态" remoteExecution status ;; *) echo "无效参数，用法为: $0 {start|stop|restart|status}" ;; esac } #定义执行的命令 function remoteExecution(){ for (( i=108 ; i<=110 ; i++ )) ; do tput setaf 2 echo ========== kafka${i} zkServer.sh $1 ================ tput setaf 9 ssh kafka${i} "source /etc/profile.d/zk.sh; zkServer.sh $1 2>/dev/null" done } #调用函数 zookeeperManger [root@kafka108 ~]# chmod +x /usr/local/sbin/manager_zk.sh 12.启动zookeeper集群 [root@kafka108 ~]# manager_zk.sh start 启动服务 ========== kafka108 zkServer.sh start ================ Starting zookeeper ... STARTED ========== kafka109 zkServer.sh start ================ Starting zookeeper ... STARTED ========== kafka110 zkServer.sh start ================ Starting zookeeper ... STARTED 13.查看集群状态 [root@kafka108 zookeeper-3.4.12]# manager_zk.sh status 查看状态 ========== kafka108 zkServer.sh status ================ Mode: follower ========== kafka109 zkServer.sh status ================ Mode: leader ========== kafka110 zkServer.sh status ================ Mode: follower 14.zk的leader选举流程 myid: 表示当前zookeeper server的server id，用于唯一表示当前的服务器。 zxid: 表示zookeeper transaction id，即zookeeper集群事务ID。 leader的选举流程如下: (1)启动kafka108和kafka109这两个节点，先检查有没有leder节点      此时2个节点进入looking状态，它们认为此时集群没有leader。 (2)若没有leader角色kafka108和kafka109这两个节点会投自己一票:      kafka108节点启动zk服务，开始发起投票 ---> [myid,zxid] --->[108, 0]      kafka109节点启动zk服务，开始发起投票 ---> [myid,zxid] --->[109, 0] (3)两个节点开始接收彼此的投票信息，并处理投票进行如下的比较:      1)优先检查zxid，即事务ID，谁的zxid较大优先成为leader;      2)如果zxid相同，那么就比较myid，较大的myid作为leader服务器;      一旦确定leader角色，此时leader服务器的状态变更为leading。      没有成为leader角色的节点，会主动变更为following状态。 (4)kafka110节点启动，先检查有没有leader，发现集群leader在上一步已经存在      server.0=10.0.0.108:2888:3888      server.1=10.0.0.109:2888:3888      server.2=10.0.0.110:2888:3888      由于leader角色已经选出来了，此时会主动变更为"following状态" (5)kafka110主从将自己的角色变更为follwer。      集群半数以上节点投票完毕就已经确定了leader角色，无需全票通过。 zookeeper基于zab协议写入数据： zookeeper原子广播协议，即Zookeeper Atomic Broadcast(简称"ZAB")。 zookeeper是通过zab协议来保证分布式事务的最终一致性。 zab工作流程如下: (1)客户端如果对follower节点发起写操作; (2)follower节点会将请求转发给leader; (3)leader生成一个新的事务并为这个事务生成一个分布式唯一的ZXID，leader将这个事物提议(propose)发送给所有的followers节点; (4)所有的follower角色开始写入数据，并将写入结果返回给leader，即响应ACK; (5)leader节点收到ACK后，需要当前集群半数以上的ACK写入成功，则提交本次事务; (6)数据写入成功则响应客户端; zookeeper的客户端常用命令: 1.连接客户端 zkCli.sh # 连接本地 zkCli.sh -server kafka108:2181 # 连接指定节点 zkCli.sh -server kafka108:2181,kafka109:2181,kafka110:2181 # 连接集群 2.zookeeper的基本使用 - 查: - ls / 查看/路径下有多少个znode。(可以理解为linux目录) - get /linux86 查看"/linux86"是否有数据。 - stat /school 查看/school的znode状态信息。 - 增: - create /linux86 在"/"路径下创建一个linux86的znode。 - create /school www.oldboyedu.com 在创建"/school"同时写入数据。 - create /school/class linux86 在"/school"路径下，创建一个名为"class"的znode，其数据为linux86。 - create -s /school/class 123 创建"/school/class"为前缀， - 改: - set /linux86 123 修改"/linux86"的znode的数据。 - 删: - delete /linux86 删除"/linux86"的znode。 - deleteall /school 递归删除"/school"下的所有znode数据。 3.kafka单点部署1.解压软件包 [root@kafka108 app]# tar xf kafka_2.12-2.5.1 2.配置环境变量 [root@kafka108 app]# vim /etc/profile.d/kafka.sh #!/bin/bash export KAFKA_HOME=/zy/app/kafka_2.12-2.5.1 export PATH=$PATH:${KAFKA_HOME}/bin [root@kafka108 app]# source /etc/profile.d/kafka.sh 3.修改配置文件 [root@kafka108 app]# vim $KAFKA_HOME/config/server.properties ... # 指定broker的唯一ID broker.id=0 ... # 指定数据目录 log.dirs=/zy/logs/kafka ... # 指定ZK的集群并指定chroot目录 zookeeper.connect=kafka108:2181,kafka109:2181,kafka110:2181/kafka 4.启动kafka服务 [root@kafka108 app]# kafka-server-start.sh $KAFKA_HOME/config/server.properties [root@kafka108 app]# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties #后台启动4.kafka集群部署1.同步kafka程序包到其他节点 [root@kafka108 ~]# data_rsync.sh /zy/app/kafka_2.12-2.5.1 [root@kafka108 ~]# data_rsync.sh /etc/profile.d/kafka.sh 2.修改其他节点的配置文件 [root@kafka109 ~]# vim $KAFKA_HOME/config/server.properties ... broker.id=1 [root@kafka110 ~]# vim $KAFKA_HOME/config/server.properties ... broker.id=2 3.各节点启动服务 [root@kafka108 ~]# for ((host_id=108;host_id<=110;host_id++)) do ssh kafka${host_id} "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties ";done5.使用kafka连接工具测试我采用的连接工具是kafka-tools6.kafka topic管理1.增 (1) 创建一个名为"oldboyedu-test02"的topic [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --create --topic oldboyedu-test02 Created topic oldboyedu-test01. (2) 创建一个名为"oldboyedu-test03"的topic，分区数为3 [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --create --topic oldboyedu-test03 --partitions 3 Created topic oldboyedu-test03. (3) 创建一个名为"oldboyedu-test04"的topic，分区数为3，副本数为2 [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --create --topic oldboyedu-test04 --partitions 3 --replication-factor 2 Created topic oldboyedu-test04. 2.查 (1) 查看topic列表 [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --list oldboyedu-test02 oldboyedu-test03 oldboyedu-test04 (2) 查看所有topic的详细信息 [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --describe (3) 查看指定topic的详细信息 [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --describe --topic oldboyedu-test03 4.改 (1) 将"oldboyedu-test02"的topic分区数增加为5个 [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --alter --topic oldboyedu-test02 --partitions 5 5.删 (1) 删除一个名为"oldboyedu-test03"的topic [root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --delete --topic oldboyedu-test037.生产者和消费者1.启动生产者 [root@kafka108 bin]# kafka-console-producer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 >n >sncj >lixiaolong 2. 启动消费者 (表示从头开始拉取数据) (1) 从头开始拉取数据 [root@kafka108 bin]# kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 --from-beginning n sncj lixiaolong (2) 表示从最新的offset拉取数据 [root@kafka108 bin]# kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 (3) kafka消费者组管理 [root@kafka108 bin]# kafka-consumer-groups.sh --bootstrap-server 10.0.0.108:9092 --list #列出所有的消费者组信息 [root@kafka108 bin]# kafka-consumer-groups.sh --bootstrap-server 10.0.0.108:9092 --describe --all-groups #列出所有消费者组的详细信息，包括偏移量，消费者ID，LEO等信息 kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning #查看内置topic的offset数据 (4) 基于配置文件指定消费者组 [root@kafka108 ~]# vim $KAFKA_HOME/config/consumer.properties ... # 指定消费者组的名称 group.id=oldboyedu-test02 [root@kafka108 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 --consumer.config $KAFKA_HOME/config/consumer.properties (5) 基于命令行参数指定消费者组 [root@kafka108 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 --consumer-property group.id=oldboyedu-test028.配置身份验证1.修改配置文件 (1) 分别在三台服务修改kafka配置文件server.properties [root@kafka108 config]# vim server.properties #增加以下内容： advertised.listeners=SASL_PLAINTEXT://10.0.0.108:9092 #这里ip地址是设置本服务器的ip，修改其他两节点需要修改成本机的ip listeners=SASL_PLAINTEXT://10.0.0.108:9092 #这里ip地址是设置本服务器的ip，修改其他两节点需要修改成本机的ip #使用的认证协议 security.inter.broker.protocol=SASL_PLAINTEXT #SASL机制 sasl.enabled.mechanisms=PLAIN sasl.mechanism.inter.broker.protocol=PLAIN #完成身份验证的类 authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer #如果没有找到ACL（访问控制列表）配置，则允许任何操作。 allow.everyone.if.no.acl.found=true #需要开启设置超级管理员, 开启以下命令 super.users=User:admin (2) 修改配置文件producer.properties [root@kafka108 config]# vim producer.properties #增加以下内容： security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN (3) 修改配置文件consumer.properties [root@kafka108 config]# vim consumer.properties #增加以下内容： security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN 2.新增配置文件 (1) 新增 kafka_server_jaas.conf配置文件 [root@kafka108 config]# vim kafka_server_jaas.conf KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="Gfgopdb@2022" user_admin="Gfgopdb@2022" user_fms="Gfgopdb@2022"; }; (2) 新增 kafka_client_jaas.conf配置文件 [root@kafka108 config]# vim kafka_client_jaas.conf KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="Gfgopdb@2022" user_admin="Gfgopdb@2022" user_fms="Gfgopdb@2022"; }; 3.修改启动脚本 (1) 修改配置文件 kafka-server-start.sh [root@kafka108 bin]# vim kafka-server-start.sh #在文件最上方增加以下内容： if [ "x$KAFKA_OPTS" ]; then export KAFKA_OPTS="-Djava.security.auth.login.config=/zy/app/kafka_2.12-2.5.1/config/kafka_server_jaas.conf" fi (2) 修改配置文件 kafka-console-producer.sh [root@kafka108 bin]# vim kafka-console-producer.sh #在文件最上方增加以下内容： if [ "x$KAFKA_OPTS" ]; then export KAFKA_OPTS="-Djava.security.auth.login.config=/zy/app/kafka_2.12-2.5.1/config/kafka_client_jaas.conf" fi 4.启动kafka (1) 要确保zookeeper集群已经启动 (2) 分别在三台服务器停掉之前启动的kafka [root@kafka108 config]# ps -ef|grep -v grep|grep kafka|awk '{print $2}'|xargs kill -9 (3) 各节点启动kafka [root@kafka108 config]# for ((host_id=108;host_id<=110;host_id++)) do ssh kafka${host_id} "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties ";done9.重新在kafka连接工具测试username="fms" password="Gfgopdb@2022" #用户密码 在连接工具里面有个JAAS Config文件里面增加以下内容： org.apache.kafka.common.security.plain.PlainLoginModule required username="fms" password="Gfgopdb@2022";

Flink standalone部署 1.准备软件包flink-1.20.0-bin-scala_2.12.tgz jdk-8u371-linux-x64.tar.gz2.准备工作目录mkdir /data/3.配置集群免密登录1.在每个节点执行如下命令，然后一直回车即可 [root@flink111 ~]# ssh-keygen -t rsa [root@flink112 ~]# ssh-keygen -t rsa [root@flink113 ~]# ssh-keygen -t rsa 2.分别将从节点的/root/.ssh/中的id_rsa.pub文件拷贝至主节点对应位置（需要重命名已避免冲突） [root@flink112 .ssh]# scp id_rsa.pub 10.0.0.111:~/.ssh/id_rsa.pub.flink-112 [root@flink113 .ssh]# scp id_rsa.pub 10.0.0.111:~/.ssh/id_rsa.pub.flink-113 3.将主节点中的id_rsa.pub⽂件添加到authorized_keys⽂件中 [root@flink111 .ssh]# cat id_rsa.pub >> authorized_keys [root@flink111 .ssh]# cat id_rsa.pub.flink-112 >> authorized_keys [root@flink111 .ssh]# cat id_rsa.pub.flink-113 >> authorized_keys 4.将authorized_keys⽂件拷⻉⾄其他节点即可 [root@flink111 .ssh]# scp authorized_keys 10.0.0.112:~/.ssh/ [root@flink111 .ssh]# scp authorized_keys 10.0.0.113:~/.ssh/ 5.然后从1号节点开始，分别ssh其他所有节点。⽬的是为了记住机器编号。也就是known_host [root@flink111 .ssh]# ssh 10.0.0.111 [root@flink111 .ssh]# ssh 10.0.0.112 [root@flink111 .ssh]# ssh 10.0.0.113 [root@flink112 .ssh]# ssh 10.0.0.111 [root@flink112 .ssh]# ssh 10.0.0.112 [root@flink112 .ssh]# ssh 10.0.0.113 [root@flink113 .ssh]# ssh 10.0.0.111 [root@flink113 .ssh]# ssh 10.0.0.112 [root@flink113 .ssh]# ssh 10.0.0.113 6.查看authorized_keys⽂件，如果⽂件⽤户组权限为读和写，去掉⽤户组写权限即可。 [root@flink111 .ssh]# chmod g-w authorized_keys [root@flink112 .ssh]# chmod g-w authorized_keys [root@flink113 .ssh]# chmod g-w authorized_keys4.配置jdk1.解压jdk tar xf jdk-8u371-linux-x64.tar.gz -C /data/ 2.配置环境变量 vim /etc/profile ... export JAVA_HOME=/data/jdk1.8.0_371 export FLINK_HOME=/data/flink-1.20.0 export PATH=$PATH:$JAVA_HOME/bin:$FLINK_HOME/bin 3.验证 [root@flink111 data]# java -version java version "1.8.0_371" Java(TM) SE Runtime Environment (build 1.8.0_371-b11) Java HotSpot(TM) 64-Bit Server VM (build 25.371-b11, mixed mode)5.部署flink1.解压flink tar xf flink-1.20.0-bin-scala_2.12.tgz -C /data/ 2.修改配置文件 cd /data/flink-1.20.0/conf/ cp config.yaml config.yaml_bak [root@flink111 conf]# cat masters localhost:8081 [root@flink111 conf]# cat workers flink111 flink112 flink113 [root@flink111 conf]# cat config.yaml env: java: opts: all: --add-exports=java.base/sun.net.util=ALL-UNNAMED --add-exports=java.rmi/sun.rmi.registry=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.file=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.parser=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.tree=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-exports=java.security.jgss/sun.security.krb5=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.io=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.lang.reflect=ALL-UNNAMED --add-opens=java.base/java.text=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED jobmanager: bind-host: 0.0.0.0 rpc: address: 10.0.0.111 port: 6123 memory: process: size: 1600m execution: failover-strategy: region taskmanager: bind-host: 0.0.0.0 host: 10.0.0.113 numberOfTaskSlots: 4 memory: process: size: 1728m parallelism: default: 1 rest: address: 10.0.0.111 bind-address: 0.0.0.0 flamegraph: enabled: true blob: storage: directory: /data/flink_bak/flink_blob 3.添加hosts vim /etc/hosts 10.0.0.111 flink111 10.0.0.112 flink112 10.0.0.113 flink113 4.复制到其他节点 scp -r /data/* 10.0.0.112:/data/ scp -r /data/* 10.0.0.113:/data/ scp /etc/profile 10.0.0.112:/etc/profile scp /etc/profile 10.0.0.113:/etc/profile scp /etc/profile 10.0.0.112:/etc/hosts scp /etc/profile 10.0.0.113:/etc/hosts 5.使环境变量生效 source /etc/profile 6.启动flink 在flink111主节点执行 start-cluster.sh 7.访问WebUI 10.0.0.111:8081 8.测试实例 nc -lk 9999 #开启9999端口 rm -rf /data/flink-1.20.0/log/* flink run /data/flink-1.20.0/examples/streaming/SocketWindowWordCount.jar --port 9999 9.打开WEB UI页面，首先看到我们有一个实例即作业在运行中，查看实例运行结果 10.停止集群 stop-cluster.sh

flink on yarn HA部署 1.准备工作目录[root@flink111 ~]# mkdir /zy/{data/{flink,hadoop,zookeeper},logs/{flink,hadoop,zookeeper},app} -p [root@flink112 ~]# mkdir /zy/{data/{flink,hadoop,zookeeper},logs/{flink,hadoop,zookeeper},app} -p [root@flink113 ~]# mkdir /zy/{data/{flink,hadoop,zookeeper},logs/{flink,hadoop,zookeeper},app} -p2.准备安装包传到/zy/app目录[root@flink111 ~]# cd /zy/app/ [root@flink111 app]# ll total 743324 -rw-r--r-- 1 root root 314015504 Jun 11 18:44 flink-1.13.6-bin-scala_2.11.tgz -rw-r--r-- 1 root root 218720521 Jun 13 16:03 hadoop-2.7.7.tar.gz -rw-r--r-- 1 root root 191753373 Jun 11 18:43 jdk-8u191-linux-x64.tar.gz -rw-r--r-- 1 root root 36667596 Jun 13 15:18 zookeeper-3.4.12.tar.gz -rw-r--r-- 1 root root 41368997 Jul 12 22:06 flink-shaded-hadoop-2-uber-2.7.5-10.0.jar #flink的依赖包3.配置集群间免密钥登陆1.在每个节点执行如下命令，然后一直回车即可 [root@flink111 ~]# ssh-keygen -t rsa [root@flink112 ~]# ssh-keygen -t rsa [root@flink113 ~]# ssh-keygen -t rsa 2.分别将从节点的/root/.ssh/中的id_rsa.pub文件拷贝至主节点对应位置（需要重命名已避免冲突） [root@flink112 .ssh]# scp id_rsa.pub 10.0.0.111:~/.ssh/id_rsa.pub.flink-112 [root@flink113 .ssh]# scp id_rsa.pub 10.0.0.111:~/.ssh/id_rsa.pub.flink-113 3.将主节点中的id_rsa.pub⽂件添加到authorized_keys⽂件中 [root@flink111 .ssh]# cat id_rsa.pub >> authorized_keys [root@flink111 .ssh]# cat id_rsa.pub.flink-112 >> authorized_keys [root@flink111 .ssh]# cat id_rsa.pub.flink-113 >> authorized_keys 4.将authorized_keys⽂件拷⻉⾄其他节点即可 [root@flink111 .ssh]# scp authorized_keys 10.0.0.112:~/.ssh/ [root@flink111 .ssh]# scp authorized_keys 10.0.0.113:~/.ssh/ 5.然后从1号节点开始，分别ssh其他所有节点。⽬的是为了记住机器编号。也就是known_host [root@flink111 .ssh]# ssh 10.0.0.111 [root@flink111 .ssh]# ssh 10.0.0.112 [root@flink111 .ssh]# ssh 10.0.0.113 [root@flink112 .ssh]# ssh 10.0.0.111 [root@flink112 .ssh]# ssh 10.0.0.112 [root@flink112 .ssh]# ssh 10.0.0.113 [root@flink113 .ssh]# ssh 10.0.0.111 [root@flink113 .ssh]# ssh 10.0.0.112 [root@flink113 .ssh]# ssh 10.0.0.113 6.查看authorized_keys⽂件，如果⽂件⽤户组权限为读和写，去掉⽤户组写权限即可。 [root@flink111 .ssh]# chmod g-w authorized_keys [root@flink112 .ssh]# chmod g-w authorized_keys [root@flink113 .ssh]# chmod g-w authorized_keys4.安装JDK1.解压软件包 [root@flink111 app]# tar xf jdk-8u191-linux-x64.tar.gz [root@flink111 app]# tar xf zookeeper-3.4.12.tar.gz [root@flink111 app]# tar xf flink-1.13.6-bin-scala_2.11.tgz [root@flink111 app]# tar xf hadoop-2.7.7.tar.gz 5.配置环境变量[root@flink111 app]# vim /etc/profile #在最下面增加以下内容: export JAVA_HOME=/zy/app/jdk1.8.0_191 export CLASSPATH=$JAVA_HOME/lib:$JRE_HOME/:lib:$CLASSPATH export ZOOKEEPER_HOME=/zy/app/zookeeper-3.4.12 export HADOOP_HOME=/zy/app/hadoop-2.7.7 export FLINK_HOME=/zy/app/flink-1.13.6 export PATH=$PATH:$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$HADOOP_HOME/sbin:$HADOOP_HOME/bin:$FLINK_HOME/bin [root@flink111 app]# source /etc/profile6.安装zookeeper1.修改配置文件zoo.cfg [root@flink111 ~]# cp /zy/app/zookeeper-3.4.12/conf/{zoo_sample,zoo}.cfg [root@flink111 ~]# vim /zy/app/zookeeper-3.4.12/conf/zoo.cfg ... # 指定数据目录 dataDir=/zy/data/zookeeper #指定日志目录 dataLogDir=/zy/logs/zookeeper ... server.0=10.0.0.111:2888:3888 server.1=10.0.0.112:2888:3888 server.2=10.0.0.113:2888:3888 2.将配置文件同步到其他节点 [root@flink111 ~]# scp /zy/app/zookeeper-3.4.12/conf/zoo.cfg 10.0.0.112:/zy/app/zookeeper-3.4.12/conf/ [root@flink111 ~]# scp /zy/app/zookeeper-3.4.12/conf/zoo.cfg 10.0.0.113:/zy/app/zookeeper-3.4.12/conf/ 3.各节点创建myid文件 [root@flink111 ~]# cd /zy/data/zookeeper/ [root@flink111 zookeeper]# echo 0 > myid [root@flink112 ~]# cd /zy/data/zookeeper/ [root@flink112 zookeeper]# echo 1 > myid [root@flink113 ~]# cd /zy/data/zookeeper/ [root@flink113 zookeeper]# echo 2 > myid 3.各节点启动zookeeper [root@flink111 zookeeper]# zkServer.sh start [root@flink112 zookeeper]# zkServer.sh start [root@flink113 zookeeper]# zkServer.sh start7.安装hadoop1.各节点准备工作目录 [root@flink111 ~]# mkdir -p /zy/data/hadoop/{tmp,journal,hdfs/{name,data,namesecondary},nm-local-dir,pid} [root@flink112 ~]# mkdir -p /zy/data/hadoop/{tmp,journal,hdfs/{name,data,namesecondary},nm-local-dir,pid} [root@flink113 ~]# mkdir -p /zy/data/hadoop/{tmp,journal,hdfs/{name,data,namesecondary},nm-local-dir,pid} 2.修改配置文件 (1) core-site.xml [root@flink111 ~]# cd /zy/app/hadoop-2.7.7/etc/hadoop [root@flink111 hadoop]# cat core-site.xml #在这里面增加以下内容 <configuration> <!-- 指定hdfs的nameservice为ns --> <property> <name>fs.defaultFS</name> <value>hdfs://ns</value> </property> <!--指定hadoop数据临时存放⽬录--> <property> <name>hadoop.tmp.dir</name> <value>/zy/data/hadoop/tmp</value> </property> <property> <name>io.file.buffer.size</name> <value>4096</value> </property> <!--指定zookeeper地址--> <property> <name>ha.zookeeper.quorum</name> <value>flink111:2181,flink112:2181,flink113:2181</value> </property> </configuration> (2) hdfs-site.xml [root@flink111 hadoop]# cat hdfs-site.xml ... <configuration> <!--指定hdfs的nameservice为ns，需要和core-site.xml中的保持⼀致 --> <property> <name>dfs.nameservices</name> <value>ns</value> </property> <!-- ns下⾯有两个NameNode，分别是nn1，nn2 --> <property> <name>dfs.ha.namenodes.ns</name> <value>nn1,nn2</value> </property> <!-- nn1的RPC通信地址 --> <property> <name>dfs.namenode.rpc-address.ns.nn1</name> <value>flink111:9000</value> </property> <!-- nn1的http通信地址 --> <property> <name>dfs.namenode.http-address.ns.nn1</name> <value>flink111:50070</value> </property> <!-- nn2的RPC通信地址 --> <property> <name>dfs.namenode.rpc-address.ns.nn2</name> <value>flink112:9000</value> </property> <!-- nn2的http通信地址 --> <property> <name>dfs.namenode.http-address.ns.nn2</name> <value>flink112:50070</value> </property> <!-- 指定NameNode的元数据在JournalNode上的存放位置 --> <property> <name>dfs.namenode.shared.edits.dir</name> <value>qjournal://flink111:8485;flink112:8485;flink113:8485/ns</value> </property> <!-- 指定JournalNode在本地磁盘存放数据的位置 --> <property> <name>dfs.journalnode.edits.dir</name> <value>/zy/data/hadoop/journal</value> </property> <!-- 开启NameNode故障时⾃动切换 --> <property> <name>dfs.ha.automatic-failover.enabled</name> <value>true</value> </property> <!-- 配置失败⾃动切换实现⽅式 --> <property> <name>dfs.client.failover.proxy.provider.ns</name> <value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value> </property> <!-- 配置隔离机制，如果ssh是默认22端⼝，value直接写sshfence即可 --> <property> <name>dfs.ha.fencing.methods</name> <value>sshfence</value> </property> <!-- 使⽤隔离机制时需要ssh免登陆 --> <property> <name>dfs.ha.fencing.ssh.private-key-files</name> <value>/root/.ssh/id_rsa</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>file:/zy/data/hadoop/hdfs/name</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>file:/zy/data/hadoop/hdfs/data</value> </property> <property> <name>dfs.replication</name> <value>2</value> </property> <!-- 在NN和DN上开启WebHDFS (REST API)功能,不是必须 --> <!--webui页面可以操作文件--> <property> <name>dfs.webhdfs.enabled</name> <value>true</value> </property> </configuration> (3) yarn-site.xml [root@flink111 hadoop]# cat yarn-site.xml ... <configuration> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> <property> <name>yarn.log-aggregation-enable</name> <value>true</value> </property> <property> <name>yarn.log-aggregation.retain-seconds</name> <value>604800</value> </property> <property> <name>yarn.resourcemanager.ha.enabled</name> <value>true</value> </property> <property> <name>yarn.resourcemanager.cluster-id</name> <value>clusterrm</value> </property> <property> <name>yarn.resourcemanager.ha.rm-ids</name> <value>rm1,rm2</value> </property> <property> <name>yarn.resourcemanager.hostname.rm1</name> <value>flink111</value> </property> <property> <name>yarn.resourcemanager.hostname.rm2</name> <value>flink112</value> </property> <property> <name>yarn.resourcemanager.zk-address</name> <value>flink111:2181,flink112:2181,flink113:2181</value> </property> <property> <name>yarn.resourcemanager.recovery.enabled</name> <value>true</value> </property> <property> <name>yarn.resourcemanager.store.class</name> <value>org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore</value> </property> <property> <name>yarn.scheduler.maximum-allocation-mb</name> <value>8196</value> </property> <property> <name>yarn.scheduler.minimum-allocation-mb</name> <value>1024</value> </property> <property> <name>yarn.nodemanager.resource.memory-mb</name> <value>56320</value> </property> <property> <name>yarn.nodemanager.resource.cpu-vcores</name> <value>64</value> </property> <property> <name>yarn.nodemanager.vmem-pmem-ratio</name> <value>4.2</value> </property> <property> <name>yarn.resourcemanager.scheduler.class</name> <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value> </property> <property> <name>yarn.scheduler.fair.preemption</name> <value>true</value> </property> <!-- 下⾯配置⽤来设置集群利⽤率的阀值， 默认值0.8f，最多可以抢占到集群所有资源的80% --> <property> <name>yarn.scheduler.fair.preemption.cluster-utilization-threshold</name> <value>1.0</value> </property> <property> <name>yarn.nodemanager.local-dirs</name> <value>/zy/data/hadoop/nm-local-dir</value> </property> </configuration> (4) hadoop-env.sh [root@flink111 hadoop]# cat hadoop-env.sh #在最下面增加以下内容 export JAVA_HOME=/zy/app/jdk1.8.0_191 export HADOOP_PID_DIR=/zy/data/hadoop/pid export HADOOP_SECURE_DN_PID_DIR=${HADOOP_PID_DIR} #新版本HADOOP_SECURE_PID_DIR=${HADOOP_PID_DIR} (5) yarn-env.sh [root@flink111 hadoop]# cat yarn-env.sh #在最下面增加以下内容 export JAVA_HOME=/zy/app/jdk1.8.0_191 export YARN_PID_DIR=/zy/data/hadoop/pid #新版本HADOOP_PID_DIR=/zy/data/hadoop/pid (6) mapred-site.xml [root@flink111 hadoop]# cp mapred-site.xml.template mapred-site.xml [root@flink111 hadoop]# cat mapred-site.xml #在这里面增加以下内容 <configuration> <property> <name>mapreduce.framework.name</name> <value>yarn</value> </property> </configuration> (7) workers [root@flink111 hadoop]# cat workers localhost #集群主机的名称: flink111 flink112 flink113 (8) /etc/hosts [root@flink111 hadoop]# cat /etc/hosts #在最下面增加以下内容 10.0.0.111 flink111 10.0.0.112 flink112 10.0.0.113 flink113 3.将工作目录和配置文件传到各节点 [root@flink111 hadoop]# scp -r /zy/app/hadoop-2.7.7 10.0.0.112:/zy/app/ [root@flink111 hadoop]# scp -r /zy/app/hadoop-2.7.7 10.0.0.113:/zy/app/ [root@flink111 hadoop]# scp /etc/hosts 10.0.0.112:/etc/hosts [root@flink111 hadoop]# scp /etc/hosts 10.0.0.113:/etc/hosts8.安装flink1.修改配置文件 (1) flink-conf.yaml [root@flink111 ~]# cd /zy/app/flink-1.13.6/conf [root@flink111 conf]# cp flink-conf.yaml flink-conf.yaml_bak [root@flink111 conf]# vim flink-conf.yaml ... jobmanager.rpc.address: flink111 #将原本的内容注释掉 #jobmanager.memory.process.size: 1600m #taskmanager.memory.process.size: 1728m #taskmanager.numberOfTaskSlots: 1 #在最下面增加以下内容: web.tmpdir: /zy/data/flink/flink-jar blob.storage.directory: /zy/data/flink/flink-blob yarn.maximum-failed-containers: 20000 taskmanager.tmp.dirs: /zy/data/flink/flink-data fs.hdfs.hadoopconf: /zy/app/hadoop-2.7.7/etc/hadoop/ fs.hdfs.hdfssite: /zy/app/hadoop-2.7.7/etc/hadoop/hdfs-site.xml high-availability: zookeeper high-availability.zookeeper.quorum: flink111:2182,flink112:2182,flink113:2182 high-availability.zookeeper.storageDir: hdfs:///flink/recovery high-availability.zookeeper.path.root: /flink jobmanager.memory.process.size: 6000m jobmanager.heap.mb: 2500 jobmanager.memory.off-heap.size: 1000m jobmanager.memory.jvm-metaspace.size: 1500m taskmanager.memory.task.heap.size: 2500m taskmanager.memory.framework.heap.size: 256m taskmanager.memory.managed.size: 200m taskmanager.memory.task.off-heap.size: 1000m taskmanager.memory.process.size: 6000m taskmanager.memory.jvm-metaspace.size: 800m taskmanager.numberOfTaskSlots: 12 jobmanager.archive.fs.dir: hdfs:///flinkHistory/completed-jobs/ historyserver.web.address: 0.0.0.0 historyserver.web.port: 49999 historyserver.archive.fs.dir: hdfs:///flinkHistory/completed-jobs/ (2) zoo.cfg [root@flink111 conf]# vim zoo.cfg #修改一下内容: dataDir=/zy/data/flink/flink-zookeeper-data clientPort=2182 server.0=10.0.0.111:2889:3889 server.1=10.0.0.112:2889:3889 server.2=10.0.0.113:2889:3889 (3) workers 或者 slaves #根据版本的不同，这里有的是workers，有的是slaves [root@flink111 conf]# cat workers localhost #集群主机的名称: flink111 flink112 flink113 2.将依赖包拷⻉⾄lib⽬录下 [root@flink111 conf]# cp /zy/app/flink-shaded-hadoop-2-uber-2.7.5-10.0.jar /zy/app/flink-1.13.6/lib/ 3.各节点创建flink data⽬录 [root@flink111 ~]# mkdir -p /zy/data/flink/{flink-data,flink-zookeeper-data,flink-jar} [root@flink112 ~]# mkdir -p /zy/data/flink/{flink-data,flink-zookeeper-data,flink-jar} [root@flink113 ~]# mkdir -p /zy/data/flink/{flink-data,flink-zookeeper-data,flink-jar} 4.将flink与data⽂件夹分发到从节点中 [root@flink111 conf]# scp -r /zy/app/flink-1.13.6 10.0.0.112:/zy/app/ [root@flink111 conf]# scp -r /zy/app/flink-1.13.6 10.0.0.113:/zy/app/ [root@flink111 conf]# scp -r /zy/data/flink/* 10.0.0.113:/zy/data/flink/ [root@flink111 conf]# scp -r /zy/data/flink/* 10.0.0.112:/zy/data/flink/ 5.启动环境 (1) 启动zookeeper，在三个节点执⾏如下命令 #已启动就跳过此步 [root@flink111 ~]# zkServer.sh start [root@flink112 ~]# zkServer.sh start [root@flink113 ~]# zkServer.sh start [root@flink111 ~]# zkServer.sh status [root@flink112 ~]# zkServer.sh status [root@flink113 ~]# zkServer.sh status (2)（首次启动）创建命名空间，在namenode主节点（flink111）执⾏如下命令。 [root@flink111 ~]# hdfs zkfc -formatZK (3) 启动journalnode，在三个节点（hdfs-site.xml journalnode配置一致）执⾏如下命令 [root@flink111 ~]# hadoop-daemon.sh start journalnode #较新版本启动hdfs --daemon start journalnode [root@flink112 ~]# hadoop-daemon.sh start journalnode [root@flink113 ~]# hadoop-daemon.sh start journalnode (4)（首次启动）格式化namenode与journalnode⽬录，在namenode主节点（flink111）执⾏如下命令。 [root@flink111 ~]# hdfs namenode -format ns (5) 启动namenode进程，在namenode主节点（flink111）执⾏如下命令 [root@flink111 ~]# hadoop-daemon.sh start namenode #较新版本启动hdfs --daemon start namenode (6)（首次启动）将元数据从namenode主节点（flink111）拷⻉⾄namenode备节点（flink112）并格式化 在namenode备节点(flink112)执⾏如下命令 [root@flink112 ~]# hdfs namenode -bootstrapStandby (7) 启动namenode进程，在namenode备节点(flink112)执⾏如下命令 [root@flink112 ~]# hadoop-daemon.sh start namenode #较新版本启动hdfs --daemon start namenode (8) 在两个namenode节点都执⾏以下命令 [root@flink111 ~]# hadoop-daemon.sh start zkfc #较新版本启动hdfs --daemon start zkfc [root@flink112 ~]# hadoop-daemon.sh start zkfc #较新版本启动hdfs --daemon start zkfc (9) 启动datanode，在各个节点执⾏如下命令 [root@flink111 ~]# hadoop-daemon.sh start datanode #较新版本启动hdfs --daemon start datanode [root@flink112 ~]# hadoop-daemon.sh start datanode [root@flink113 ~]# hadoop-daemon.sh start datanode (10) 启动resourcemanager，在主节点（flink111）和备节点（flink112）执⾏如下命令 [root@flink111 ~]# yarn-daemon.sh start resourcemanager #较新版本yarn --daemon start resourcemanager [root@flink112 ~]# yarn-daemon.sh start resourcemanager (11) 启动nodemanager，在各个节点执⾏如下命令 [root@flink111 ~]# yarn-daemon.sh start nodemanager #较新版本yarn --daemon start nodemanager [root@flink112 ~]# yarn-daemon.sh start nodemanager [root@flink113 ~]# yarn-daemon.sh start nodemanager (12) 启动flink，在主节点（flink111）执⾏如下命令 [root@flink111 ~]# start-zookeeper-quorum.sh 6. 配置C:\Windows\System32\drivers\etc\hosts文件 10.0.0.111 flink111 10.0.0.112 flink112 10.0.0.113 flink113 7.在浏览器中键⼊flink111:8088查看yarn application状态 8.停止环境 停止flink，在主节点（flink111）执⾏如下命令 stop-zookeeper-quorum.sh 关闭hadoop，在主节点（flink111）执⾏如下命令 stop-all.sh 关闭zookeeper，各个节点执⾏如下命令 zkServer.sh stop jps查是否有未关闭的相关进程，有则kill掉 9.新增session nohup yarn-session.sh -nm loca_test > /zy/data/flink/flink-out/loca_test.out & 10.删除 yarn application -kill application_1675914412308_0332 11.查看 yarn application -list 12.清空所有session yarn resourcemanager -format-state-store #主节点执行后会重启resourcemanager，8088端口会转移到备节点。之后备节点执行一次即可转移回原本的。 13.新版本中遇到创建会话失败： export HADOOP_CLASSPATH=`hadoop classpath`