Flink 1.17教程：集群搭建、运行模式（standalone/yarn/k8s）及历史服务器

朴灿烈づ我的快乐病毒、 2023-10-15 14:37 55阅读 0赞

## 集群角色 ##

![img][]

## 集群启动 ##

> 如果是部署在本地，本地访问，无需进行任何配置，直接启动即可。
> 
> 如果是部署在服务器，需要远程访问，则需要将`flink.conf`中的localhost修改为服务器IP地址或是`0.0.0.0`

<table> 
 <thead> 
  <tr> 
   <th><strong>节点服务器</strong></th> 
   <th><strong>hadoop102</strong></th> 
   <th><strong>hadoop103</strong></th> 
   <th><strong>hadoop104</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>角色</strong></td> 
   <td>JobManagerTaskManager</td> 
   <td>TaskManager</td> 
   <td>TaskManager</td> 
  </tr> 
 </tbody> 
</table>

[atguigu@node001 module]$ cd flink
    [atguigu@node001 flink]$ cd flink-1.17.0/
    [atguigu@node001 flink-1.17.0]$ bin/start-cluster.sh
    Starting cluster.
    Starting standalonesession daemon on host node001.
    Starting taskexecutor daemon on host node001.
    Starting taskexecutor daemon on host node002.
    Starting taskexecutor daemon on host node003.
    [atguigu@node001 flink-1.17.0]$ jpsall 
    ================ node001 ================
    3408 Jps
    2938 StandaloneSessionClusterEntrypoint
    3276 TaskManagerRunner
    ================ node002 ================
    2852 TaskManagerRunner
    2932 Jps
    ================ node003 ================
    2864 TaskManagerRunner
    2944 Jps
    [atguigu@node001 flink-1.17.0]$

![img][img 1]

## WebUI提交作业 ##

### 打jar包maven插件配置 ###

<build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.4</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers combine.children="append">
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer">
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

![img][img 2]

com.atguigu.wc.WordCountStreamUnboundedDemo

![img][img 3]

![img][img 4]

![img][img 5]

![img][img 6]

## 命令行提交作业 ##

`bin/flink run -m node001:8081 -c com.atguigu.wc.WordCountStreamUnboundedDemo ../jar/FlinkTutorial-1.17-1.0-SNAPSHOT.jar`

连接成功
    Last login: Fri Jun 16 14:44:01 2023 from 192.168.10.1
    [atguigu@node001 ~]$ cd /opt/module/flink/flink-1.17.0/
    
    [atguigu@node001 flink-1.17.0]$ cd bin
    
    [atguigu@node001 bin]$ ./start-cluster.sh 
    Starting cluster.
    Starting standalonesession daemon on host node001.
    Starting taskexecutor daemon on host node001.
    Starting taskexecutor daemon on host node002.
    Starting taskexecutor daemon on host node003.
    
    [atguigu@node001 bin]$ jpsall 
    ================ node001 ================
    2723 TaskManagerRunner
    2855 Jps
    2380 StandaloneSessionClusterEntrypoint
    ================ node002 ================
    2294 TaskManagerRunner
    2367 Jps
    ================ node003 ================
    2292 TaskManagerRunner
    2330 Jps
    
    [atguigu@node001 bin]$ cd ..
             
    [atguigu@node001 flink-1.17.0]$ bin/flink run -m node001:8081 -c com.atguigu.wc.WordCountStreamUnboundedDemo ../jar/FlinkTutorial-1.17-1.0-SNAPSHOT.jar 
    Job has been submitted with JobID 59ae9d6532523b0c48cdb8b6c9105356

![img][img 7]

![img][img 8]

![img][img 9]

## 部署模式介绍 ##

在一些应用场景中，对于集群资源分配和占用的方式，可能会有特定的需求。Flink为各种场景提供了不同的部署模式，主要有以下三种：会话模式（Session Mode）、单作业模式（Per-Job Mode）、应用模式（Application Mode）。

它们的区别主要在于：集群的生命周期以及资源的分配方式；以及应用的main方法到底在哪里执行——客户端（Client）还是JobManager。

## Standalone运行模式 ##

独立模式是独立运行的，不依赖任何外部的资源管理平台；当然独立也是有代价的：如果资源不足，或者出现故障，没有自动扩展或重分配资源的保证，必须手动处理。所以独立模式一般只用在开发测试或作业非常少的场景下。

精简版脚本：

bin/standalone-job.sh start --job-classname com.atguigu.wc.WordCountStreamUnboundedDemo
    
    bin/taskmanager.sh start
    
    bin/taskmanager.sh stop
    
    bin/standalone-job.sh stop

详细展示版：

[atguigu@node001 ~]$ cd /opt/module/flink/flink-1.17.0/bin
    
    [atguigu@node001 bin]$ ./stop-cluster.sh 
    Stopping taskexecutor daemon (pid: 2723) on host node001.
    Stopping taskexecutor daemon (pid: 2294) on host node002.
    Stopping taskexecutor daemon (pid: 2292) on host node003.
    Stopping standalonesession daemon (pid: 2380) on host node001.
    
    [atguigu@node001 bin]$ jpsall 
    ================ node001 ================
    5120 Jps
    ================ node002 ================
    3212 Jps
    ================ node003 ================
    3159 Jps
    
    [atguigu@node001 bin]$ ls
    bash-java-utils.jar  flink             historyserver.sh          kubernetes-session.sh      sql-client.sh      start-cluster.sh           stop-zookeeper-quorum.sh  zookeeper.sh
    config.sh            flink-console.sh  jobmanager.sh             kubernetes-taskmanager.sh  sql-gateway.sh     start-zookeeper-quorum.sh  taskmanager.sh
    find-flink-home.sh   flink-daemon.sh   kubernetes-jobmanager.sh  pyflink-shell.sh           standalone-job.sh  stop-cluster.sh            yarn-session.sh
    
    [atguigu@node001 bin]$ cd ../lib/
    
    [atguigu@node001 lib]$ ls
    flink-cep-1.17.0.jar              flink-dist-1.17.0.jar        flink-table-api-java-uber-1.17.0.jar   FlinkTutorial-1.17-1.0-SNAPSHOT.jar  log4j-core-2.17.1.jar
    flink-connector-files-1.17.0.jar  flink-json-1.17.0.jar        flink-table-planner-loader-1.17.0.jar  log4j-1.2-api-2.17.1.jar             log4j-slf4j-impl-2.17.1.jar
    flink-csv-1.17.0.jar              flink-scala_2.12-1.17.0.jar  flink-table-runtime-1.17.0.jar         log4j-api-2.17.1.jar
    
    [atguigu@node001 lib]$ cd ../
    
    [atguigu@node001 flink-1.17.0]$ bin/standalone-job.sh start --job-classname com.atguigu.wc.WordCountStreamUnboundedDemo
    Starting standalonejob daemon on host node001.
    
    [atguigu@node001 flink-1.17.0]$ jpsall 
    ================ node001 ================
    5491 StandaloneApplicationClusterEntryPoint
    5583 Jps
    ================ node002 ================
    3326 Jps
    ================ node003 ================
    3307 Jps
    
    [atguigu@node001 flink-1.17.0]$ bin/taskmanager.sh 
    Usage: taskmanager.sh (start|start-foreground|stop|stop-all)
    
    [atguigu@node001 flink-1.17.0]$ bin/taskmanager.sh start
    Starting taskexecutor daemon on host node001.
    
    [atguigu@node001 flink-1.17.0]$ jpsall 
    ================ node001 ================
    5491 StandaloneApplicationClusterEntryPoint
    5995 Jps
    5903 TaskManagerRunner
    ================ node002 ================
    3363 Jps
    ================ node003 ================
    3350 Jps
    
    [atguigu@node001 flink-1.17.0]$ bin/taskmanager.sh stop
    Stopping taskexecutor daemon (pid: 5903) on host node001.
    
    [atguigu@node001 flink-1.17.0]$ bin/standalone-job.sh stop
    No standalonejob daemon (pid: 5491) is running anymore on node001.
    
    [atguigu@node001 flink-1.17.0]$ xcall jps
    =============== node001 ===============
    6682 Jps
    =============== node002 ===============
    3429 Jps
    =============== node003 ===============
    3419 Jps

## YARN运行模式\_环境准备 ##

YARN上部署的过程是：客户端把Flink应用提交给Yarn的ResourceManager，Yarn的ResourceManager会向Yarn的NodeManager申请容器。在这些容器上，Flink会部署JobManager和TaskManager的实例，从而启动集群。Flink会根据运行在JobManger上的作业所需要的Slot数量动态分配TaskManager资源。

[atguigu@node001 flink-1.17.0]$ source /etc/profile.d/my_env.sh 
    [atguigu@node001 flink-1.17.0]$ myhadoop.sh s
    Input Args Error...
    [atguigu@node001 flink-1.17.0]$ myhadoop.sh start
     ================ 启动 hadoop集群 ================
     ---------------- 启动 hdfs ----------------
    Starting namenodes on [node001]
    Starting datanodes
    Starting secondary namenodes [node003]
     --------------- 启动 yarn ---------------
    Starting resourcemanager
    Starting nodemanagers
     --------------- 启动 historyserver ---------------
    [atguigu@node001 flink-1.17.0]$ jpsall 
    ================ node001 ================
    9200 JobHistoryServer
    8416 NameNode
    8580 DataNode
    9284 Jps
    8983 NodeManager
    ================ node002 ================
    3892 ResourceManager
    3690 DataNode
    4365 Jps
    4015 NodeManager
    ================ node003 ================
    3680 DataNode
    3778 SecondaryNameNode
    3911 NodeManager
    4044 Jps
    [atguigu@node001 flink-1.17.0]$

## YARN运行模式\_会话模式 ##

这段命令是用于启动 Apache Flink 的 YARN 会话（session）的脚本，如下是每个选项和参数的含义：

*  `yarn-session.sh`：这是 Apache Flink 提供的用于在 YARN 上启动会话的脚本。
 *  `-d`：这是一个选项，表示以分离模式（detached mode）启动会话。在分离模式下，会话将在后台运行，并且脚本会立即返回。
 *  `-nm test`：这是另一个选项，用于指定会话的名称。在这个例子中，会话的名称被设置为 “test”。

综合起来，该命令的目的是在 YARN 上启动一个名为 “test” 的 Apache Flink 会话，并以分离模式运行。启动后，该会话将在后台运行，并且命令行提示符会立即返回，允许您继续执行其他操作。

[atguigu@node001 bin]$ ./yarn-session.sh --help
    [atguigu@node001 bin]$ ./yarn-session.sh
    [atguigu@node001 bin]$ ./yarn-session.sh -d -nm test

![img][img 10]

![img][img 11]

![img][img 12]

![img][img 13]

## YARN运行模式\_会话模式的停止 ##

### 单作业模式部署 ###

在YARN环境中，由于有了外部平台做资源调度，所以我们也可以直接向YARN提交一个单独的作业，从而启动一个Flink集群。

![img][img 14]

停止job：

![img][img 15]

## YARN运行模式\_单作业模式 ##

### 单作业模式部署 ###

（1）执行命令提交作业

![img][img 16]

## YARN运行模式\_应用模式 ##

应用模式同样非常简单，与单作业模式类似，直接执行

flink run-application

命令即可。如：

bin/flink run-application -t yarn-application -c com.atguigu.wc.WordCountStreamUnboundedDemo ./FlinkTutorial-1.17-1.0-SNAPSHOT.jar

[atguigu@node001 flink-1.17.0]$ bin/flink run-application -t yarn-application -c com.atguigu.wc.WordCountStreamUnboundedDemo ./FlinkTutorial-1.17-1.0-SNAPSHOT.jar 
    
    SLF4J: Class path contains multiple SLF4J bindings.
    SLF4J: Found binding in [jar:file:/opt/module/flink/flink-1.17.0/lib/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    SLF4J: Found binding in [jar:file:/opt/module/hadoop/hadoop-3.1.3/share/hadoop/common/lib/slf4j-log4j12-1.7.25.jar!/org/slf4j/impl/StaticLoggerBinder.class]
    SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
    SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
    2023-06-19 14:31:05,693 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Found Yarn properties file under /tmp/.yarn-properties-atguigu.
    2023-06-19 14:31:05,693 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                [] - Found Yarn properties file under /tmp/.yarn-properties-atguigu.
    2023-06-19 14:31:06,142 WARN  org.apache.flink.yarn.configuration.YarnLogConfigUtil        [] - The configuration directory ('/opt/module/flink/flink-1.17.0/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
    2023-06-19 14:31:06,632 INFO  org.apache.hadoop.yarn.client.RMProxy                        [] - Connecting to ResourceManager at node002/192.168.10.102:8032
    2023-06-19 14:31:07,195 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar

![img][img 17]![img][img 18]

jar存放在hdfs的情况：

[atguigu@node001 flink-1.17.0]$ bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs://node001:8020/flink-dist" -c com.atguigu.wc.WordCountStreamUnboundedDemo hdfs://node001:8020/flink-jars/FlinkTutorial-1.17-1.0-SNAPSHOT.jar

## K8S 运行模式 ##

容器化部署是如今业界流行的一项技术，基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes（k8s），而Flink也在最近的版本中支持了k8s部署模式。基本原理与YARN是类似的，具体配置可以参见官网说明，这里我们就不做过多讲解了。

## 历史服务器History Server ##

运行 Flink job 的集群一旦停止，只能去 yarn 或本地磁盘上查看日志，不再可以查看作业挂掉之前的运行的 Web UI，很难清楚知道作业在挂的那一刻到底发生了什么。如果我们还没有 Metrics 监控的话，那么完全就只能通过日志去分析和定位问题了，所以如果能还原之前的 Web UI，我们可以通过 UI 发现和定位一些问题。

Flink提供了历史服务器，用来在相应的 Flink 集群关闭后查询已完成作业的统计信息。我们都知道只有当作业处于运行中的状态，才能够查看到相关的WebUI统计信息。通过 `History Server` 我们才能查询这些已完成作业的统计信息，无论是正常退出还是异常退出。

此外，它对外提供了 REST API，它接受 HTTP 请求并使用 JSON 数据进行响应。Flink 任务停止后，JobManager 会将已经完成任务的统计信息进行存档，History Server 进程则在任务停止后可以对任务统计信息进行查询。比如：最后一次的 Checkpoint、任务运行时的相关配置。

![img][img 19]

启动historyserver：

[atguigu@node001 flink-1.17.0]$ bin/historyserver.sh start
    
    Starting historyserver daemon on host node001.
    
    [atguigu@node001 flink-1.17.0]$ bin/flink run -t yarn-per-job -d -c com.atguigu.wc.WordCountStreamUnboundedDemo ../jar/FlinkTutorial-1.17-1.0-SNAPSHOT.jar
    
    SLF4J: Class path contains multiple SLF4J bindings.
    SLF4J: Found binding in [jar:file:/opt/module/flink/flink-1.17.0/lib/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]

![img][img 20]

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

links：

[https://upward.blog.csdn.net/article/details/131215329][https_upward.blog.csdn.net_article_details_131215329]

[img]: https://img-blog.csdnimg.cn/f3054d8294d6412b85ee53d037322d75.png
[img 1]: https://img-blog.csdnimg.cn/93566c0f110e4d9f85e1bb4ebcf478c2.png
[img 2]: https://img-blog.csdnimg.cn/e3a03f333077468284e2312bef830482.png
[img 3]: https://img-blog.csdnimg.cn/94684d1adc5d421a91d9ce9bb1e0d8d8.png
[img 4]: https://img-blog.csdnimg.cn/96a67b3ccf644a589c66d266a2fc8d37.png
[img 5]: https://img-blog.csdnimg.cn/a8e5aac47ebc4c6fa0caf1e7d00fcc89.png
[img 6]: https://img-blog.csdnimg.cn/2183270f2ad44bc883a0e02d884ad0fe.png
[img 7]: https://img-blog.csdnimg.cn/02d51068c11140ae9e40a6f4e539fc64.png
[img 8]: https://img-blog.csdnimg.cn/8146fbdff0ca416192d2130dae7864b2.png
[img 9]: https://img-blog.csdnimg.cn/6f22d66802294df7994c7c0e8cc18fbf.png
[img 10]: https://img-blog.csdnimg.cn/beae5c33308542509df8a644a2399ecd.png
[img 11]: https://img-blog.csdnimg.cn/1832ca1d680e41f690c42fcb92d29532.png
[img 12]: https://img-blog.csdnimg.cn/c53c211058ad452089977d106b6aa1dd.png
[img 13]: https://img-blog.csdnimg.cn/fef70eda71d54b518b91b168bd97b863.png
[img 14]: https://img-blog.csdnimg.cn/538fe9ee302d4c6693f3abda89fff2ca.png
[img 15]: https://img-blog.csdnimg.cn/4e4c5e1ab3424f779db34ea7c2266d25.png
[img 16]: https://img-blog.csdnimg.cn/a5189dbdadb54db1b0428fe0681e8256.png
[img 17]: https://img-blog.csdnimg.cn/7253d598c6b24b61be48d245b146737a.png
[img 18]: https://img-blog.csdnimg.cn/482dc38ab7b64870823338634ae47cd8.png
[img 19]: https://img-blog.csdnimg.cn/2dca27de247942af95863139ebcdada4.png
[img 20]: https://img-blog.csdnimg.cn/31d1610f42704ab3b755fb72ece361e8.png
[https_upward.blog.csdn.net_article_details_131215329]: https://upward.blog.csdn.net/article/details/131215329