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

「爱情、让人受尽委屈。」 2022-12-29 09:28 116阅读 0赞

**目录**

1.前置准备

2.搭建Standalone模式的Flink

一、集群规划

二、环境搭建

2.1 下载并解压

2.2 配置环境变量

2.3 修改配置

2.4 分发

三、启动集群

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

# 2.搭建Standalone模式的Flink #

## 一、集群规划 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

## 二、环境搭建 ##

### 2.1 下载并解压 ###

下载 `Flink` 安装包，这里我下载的是`flink-1.10.1-bin-scala_2.11.tgz`。下载地址：https://flink.apache.org/zh/downloads.html

# 解压
    [xiaokang@hadoop01 ~]$ tar -zxvf flink-1.10.1-bin-scala_2.11.tgz -C /opt/software/

### 2.2 配置环境变量 ###

[xiaokang@hadoop01 ~]$ sudo vim /etc/profile.d/env.sh

在原来基础上更新配置环境变量：

export FLINK_HOME=/opt/software/flink-1.10.1
    export PATH=${JAVA_HOME}/bin:${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:${ZOOKEEPER_HOME}/bin:${HIVE_HOME}/bin:${ZEPPELIN_HOME}/bin:${HBASE_HOME}/bin:${SQOOP_HOME}/bin:${FLUME_HOME}/bin:${PYTHON_HOME}/bin:${SCALA_HOME}/bin:${MAVEN_HOME}/bin:${GRADLE_HOME}/bin:${KAFKA_HOME}/bin:${SPARK_HOME}/bin:${FLINK_HOME}/bin:$PATH

使得配置的环境变量立即生效：

[xiaokang@hadoop01 ~]$ source /etc/profile.d/env.sh

### 2.3 修改配置 ###

**1.进入 `${FLINK_HOME}/conf` 目录下，修改`flink-conf.yaml`文件**

jobmanager.rpc.address: hadoop01
    # 将已完成的作业上传到此目录中，让任务历史服务器进行监控
    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

除了上面的配置之外，Flink 还可以使用以下可选参数来优化集群性能：

*  **jobmanager.heap.size**：JobManager 的 JVM 堆内存大小，默认为 1024m 。
 *  **taskmanager.heap.size**：Taskmanager 的 JVM 堆内存大小，默认为 1024m 。
 *  **taskmanager.numberOfTaskSlots**：Taskmanager 上 slots 的数量，通常设置为 CPU 核心的数量，或其一半。
 *  **parallelism.default**：任务默认的并行度。
 *  **io.tmp.dirs**：存储临时文件的路径，如果没有配置，则默认采用服务器的临时目录，如 LInux 的 `/tmp` 目录。

更多配置可以参考 Flink 的官方手册：https://ci.apache.org/projects/flink/flink-docs-release-1.9/ops/config.html

**2.修改 masters 文件，指定JobManager节点**

hadoop01:8081

**3.修改 slaves 文件，添加TaskManager节点**

hadoop01
    hadoop02
    hadoop03

**作业归档需要记录在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.4 分发 ###

[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--------
    19232 DFSZKFailoverController
    20722 HistoryServer
    20387 TaskManagerRunner
    19348 JobHistoryServer
    20756 Jps
    18805 DataNode
    18502 QuorumPeerMain
    19435 NodeManager
    19980 StandaloneSessionClusterEntrypoint
    19021 JournalNode
    18687 NameNode
    --------hadoop02--------
    11890 NodeManager
    11571 JournalNode
    11683 DFSZKFailoverController
    11300 QuorumPeerMain
    12597 Jps
    11801 ResourceManager
    11387 NameNode
    11468 DataNode
    12542 TaskManagerRunner
    --------hadoop03--------
    10305 QuorumPeerMain
    11218 TaskManagerRunner
    10484 JournalNode
    10612 ResourceManager
    11272 Jps
    10731 NodeManager
    10381 DataNode

### 4.2 Web UI查看 ###

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

![图片][333d6813a40dadd8fa4c7ca34d454b36.png]

![图片][1c991b5c532f5fb1f03f85b79390d389.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

![图片][36f0fdb38a424ddd3041a95bdccb2a06.png]

输入数据以及结果：

![图片][9848f59ca6f3d02d4915942dd3ed98d3.png]

![图片][42887c679704a73586cb445b31d9ad36.png]

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

![图片][8e80c928e59e71e37fb1ed87df8e7c08.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221120/93840138262849c79c756e6c267c390e.png
[333d6813a40dadd8fa4c7ca34d454b36.png]: /images/20221120/eb374fadd5c74114aefb0d67f52ea32a.png
[1c991b5c532f5fb1f03f85b79390d389.png]: /images/20221120/b208f83dbd2a4beaae328399db88cb25.png
[36f0fdb38a424ddd3041a95bdccb2a06.png]: /images/20221120/1b55bcc37b284d37a509314762e32085.png
[9848f59ca6f3d02d4915942dd3ed98d3.png]: /images/20221120/b02ac71d13e040729536ce3a3aae5b5b.png
[42887c679704a73586cb445b31d9ad36.png]: /images/20221120/3b927aaea5c64ade955a550d5123f859.png
[8e80c928e59e71e37fb1ed87df8e7c08.png]: /images/20221120/f3259c0c60fd421195ac3982fbc4f8d1.png