4.1.5 Flink-流处理框架-HA-Flink集群环境搭建(Standalone模式)

以你之姓@ 2022-12-29 09:28 122阅读 0赞

**目录**

1.前置准备

2.安装HA-Flink集群环境搭建(Standalone模式)

一、集群规划

二、集群配置

2.1 flink-conf.yaml

2.2 masters

2.3 slaves

2.4 上传hadoop依赖包

2.5 分发

三、启动集群

3.1 启动ha-hadoop集群

3.2 启动Flink集群和任务历史服务器

四、查看集群

4.1 jps进程查看

4.2 Web UI查看

五、提交任务

--------------------

# 1.前置准备 #

CentOS7、jdk1.8、flink-1.10.1、hadoop-2.7.7、zookeeper-3.5.7

# 2.安装HA-Flink集群环境搭建(Standalone模式) #

## 一、集群规划 ##

![图片][8ffa8b55ba5acd38b8b4c5a96a259d66.png]

## 二、集群配置 ##

### 2.1 flink-conf.yaml ###

# 配置使用zookeeper来开启高可用模式
    high-availability: zookeeper
    # 配置zookeeper的地址，采用zookeeper集群时，可以使用逗号来分隔多个节点地址
    high-availability.zookeeper.quorum: hadoop01:2181,hadoop02:2181,hadoop03:2181
    # 在zookeeper上存储flink集群元信息的路径
    high-availability.zookeeper.path.root: /ha-flink
    # 持久化存储JobManager元数据的地址，zookeeper上存储的只是指向该元数据的指针信息
    high-availability.storageDir: hdfs://hacluster:8020/flink/recovery
    # 将已完成的作业上传到此目录中，让任务历史服务器进行监控
    jobmanager.archive.fs.dir: hdfs://hacluster:8020/flink-jobhistory
    historyserver.web.address: hadoop01
    historyserver.web.port: 18082
    # 任务历史服务器监控目录中已存档的作业
    historyserver.archive.fs.dir: hdfs://hacluster:8020/flink-jobhistory
    historyserver.web.refresh-interval: 10000

### 2.2 masters ###

hadoop01:8081
    hadoop02:8081

### 2.3 slaves ###

hadoop01
    hadoop02
    hadoop03

### 2.4 上传hadoop依赖包 ###

**作业归档需要记录在hdfs上，但是当前版本的flink把hadoop的一些依赖删除了，需要手动将jar包放到lib目录下 ，这里我用的是`flink-shaded-hadoop-2-uber-2.7.5-10.0.jar`**

jar包(hadoop依赖包和wordcount依赖包)给大家提供下载地址：https://lanzous.com/b02b6wnfa，密码:2oby

### 2.5 分发 ###

[xiaokang@hadoop01 ~]$ distribution.sh /opt/software/flink-1.10.1

## 三、启动集群 ##

### 3.1 启动ha-hadoop集群 ###

[xiaokang@hadoop01 ~]$ ha-hadoop.sh start
    # 创建作业归档目录
    [xiaokang@hadoop01 ~]$ hdfs dfs -mkdir /flink-jobhistory

### 3.2 启动Flink集群和任务历史服务器 ###

[xiaokang@hadoop01 ~]$ start-cluster.sh
    [xiaokang@hadoop01 ~]$ historyserver.sh start

## 四、查看集群 ##

### 4.1 jps进程查看 ###

[xiaokang@hadoop01 ~]$ call-cluster.sh jps
    --------hadoop01--------
    10369 QuorumPeerMain
    11297 NodeManager
    12241 TaskManagerRunner
    10885 JournalNode
    10551 NameNode
    12599 Jps
    12538 HistoryServer
    11083 DFSZKFailoverController
    11211 JobHistoryServer
    10669 DataNode
    11823 StandaloneSessionClusterEntrypoint
    --------hadoop02--------
    8977 TaskManagerRunner
    7459 QuorumPeerMain
    7956 ResourceManager
    7542 NameNode
    7623 DataNode
    8616 StandaloneSessionClusterEntrypoint
    9066 Jps
    7821 DFSZKFailoverController
    7726 JournalNode
    8047 NodeManager
    --------hadoop03--------
    7456 QuorumPeerMain
    7636 JournalNode
    7764 ResourceManager
    7878 NodeManager
    8345 TaskManagerRunner
    8410 Jps
    7532 DataNode

### 4.2 Web UI查看 ###

两个 JobManager 和 任务历史服务器的端口号分别为 `8081` 、`8081`和 `18082`，界面应该如下：

![图片][7d4a02a2cf17adc99ed36be7d9c1e482.png]

![图片][2dd08b04064e9c6e5435942b99eff30c.png]

![图片][2a1f274e46ef6ff920a3844a7dcfcc7b.png]

## 五、提交任务 ##

流计算词频统计案例源码：

package cool.xiaokang.wordcount
    
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    
    /**
     * TODO: Flink流计算WordCount
     *
     * @author: xiaokang
     * @date: 2020/6/8 14:19
     *      
     */
    object FlinkStreamingWordCount1 {
      def main(args: Array[String]): Unit = {
        if(args==null || args.length!=2){
          println("缺少参数，使用方法：flink run FlinkStreamingWordCount-1.0.jar -c cool.xiaokang.wordcount.FlinkStreamingWordCount hadoop 1124")
          System.exit(1)
        }
        val host=args(0)
        val port=args(1)
        //1.初始化流计算环境
        val streamEnv:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        //2.导入隐式转换
        import org.apache.flink.streaming.api.scala._
        //3.读取数据(Socket流)
        val lines:DataStream[String] = streamEnv.socketTextStream(host, port.toInt)
        //4.转换和处理数据并打印结果
        val result=lines.flatMap(_.split(" "))
          .map((_,1))
          .keyBy(0) //分组算子 0或1代表前面的DataStream[(String,Int)]的下标，0代表单词，1代表词频
          .sum(1) //聚合累加算子
        result.print("Result:")
        //5.启动流计算程序
        streamEnv.execute("FlinkSteamWordCount-微信公众号：小康新鲜事儿")
      }
    }

提交作业：

# 1.虚拟机内开启端口
    [xiaokang@hadoop01 ~]$ nc -lk 1124
    # 2.提交作业
    [xiaokang@hadoop01 ~]$ flink run -c cool.xiaokang.wordcount.FlinkStreamingWordCount ~/FlinkStreamingWordCount-1.0.jar hadoop01 1124

![图片][b4f2986974c8209647b58d5159daf396.png]

输入数据以及结果：

![图片][3fb6ffebb68d97ea1d1f688c8470e422.png]

![图片][60ecd9adbb036689ba4421a76cc550c7.png]

运行过程中将正在服务的JobManager给kill掉，测试是否高可用

[xiaokang@hadoop01 ~]$ kill -9 11641

此时hadoop01的8081无法访问，hadoop02会进行接管（重新提交刚才被中断的作业），这个过程需要稍等一会儿

再次输入数据后可以从结果看出是一个新作业：

![图片][7f495082333b00a605f74f02cddae89e.png]

![图片][fc87fa44a295b7398887b70199116a5b.png]

结束任务后可以在任务历史服务器WebUI中进行查看：

![图片][ea3ea658521a65f1f24976a3891ee233.png]

[8ffa8b55ba5acd38b8b4c5a96a259d66.png]: /images/20221120/9502ecf338a74bb9a82d22eaa636d1e9.png
[7d4a02a2cf17adc99ed36be7d9c1e482.png]: /images/20221120/98adfae4497a4173992cf90b682591ba.png
[2dd08b04064e9c6e5435942b99eff30c.png]: /images/20221120/a7a74b4c47bd49b8be593b72ed5963e3.png
[2a1f274e46ef6ff920a3844a7dcfcc7b.png]: /images/20221120/e14632c3ec5f4ce0b83757a68702744e.png
[b4f2986974c8209647b58d5159daf396.png]: /images/20221120/bc316308c56d4db0a50f1be5efd25fde.png
[3fb6ffebb68d97ea1d1f688c8470e422.png]: /images/20221120/068368f285804c57bb46463f4b905d1b.png
[60ecd9adbb036689ba4421a76cc550c7.png]: /images/20221120/9fbe38d8d99c411d88311bf611e534b3.png
[7f495082333b00a605f74f02cddae89e.png]: /images/20221120/b5b82ebe3f4941409dd066060246e4dd.png
[fc87fa44a295b7398887b70199116a5b.png]: /images/20221120/722b643f96b24c7791d54f99e1e25125.png
[ea3ea658521a65f1f24976a3891ee233.png]: /images/20221120/a1727c0e0c824a1cb0c158f42a4c1188.png