【Flink入门】centos上flink安装部署(standalone模式)

待我称王封你为后i 2021-09-25 02:26 1096阅读 0赞

### 目录 ###

*  基本概念
 *  安装部署
 *   *  下载flink安装包
     *  单机 Standalone 的方式运行 Flink
     *  常用配置介绍
     *  日志的查看和配置
     *  进一步探索

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

# 基本概念 #

运行 Flink 应用其实非常简单，但是在运行 Flink 应用之前，还是有必要了解 Flink 运行时的各个组件，因为这涉及到 Flink 应用的配置问题。图 1 所示，这是用户用 DataStream API 写的一个数据处理程序。可以看到，在一个 DAG 图中不能被 Chain 在一起的 Operator 会被分隔到不同的 Task 中，也就是说 Task 是 Flink 中资源调度的最小单位。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70]

图 1. Parallel Dataflows

如下图图 2 所示，Flink 实际运行时包括两类进程：  
JobManager（又称为 JobMaster）：协调 Task 的分布式执行，包括调度 Task、协调创 Checkpoint 以及当 Job failover 时协调各个 Task 从 Checkpoint 恢复等。

TaskManager（又称为 Worker）：执行 Dataflow 中的 Tasks，包括内存 Buffer 的分配、Data Stream 的传递等。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 1]

图 2 . Flink Runtime 架构图

图 3 所示，Task Slot 是一个 TaskManager 中的最小资源分配单位，一个 TaskManager 中有多少个 Task Slot 就意味着能支持多少并发的 Task 处理。需要注意的是，一个 Task Slot 中可以执行多个 Operator，一般这些 Operator 是能被 Chain 在一起处理的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 2]

图 3 . Process

# 安装部署 #

先在centos上安装部署Java 8

## 下载flink安装包 ##

cd /opt/tools
    
    # 下载flink1.12.0版本
    wget http://mirrors.estointernet.in/apache/flink/flink-1.12.0/flink-1.12.0-bin-scala_2.11.tgz
    
    # 解压压缩包
    tar -zxvf flink-1.12.0-bin-scala_2.11.tgz -C /opt/modules/

## 单机 Standalone 的方式运行 Flink ##

最简单的运行 Flink 应用的方法就是以单机 Standalone 的方式运行。  
启动集群：

./bin/start-cluster.sh

打开 http://127.0.0.1:8081/ 就能看到 Flink 的 Web 界面。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 3]

尝试提交 Word Count 任务：

./bin/flink run examples/streaming/WordCount.jar

大家可以自行探索 Web 界面中展示的信息，比如，我们可以看看 TaskManager 的 stdout 日志，就可以看到 Word Count 的计算结果。  
我们还可以尝试通过“–input”参数指定我们自己的本地文件作为输入，然后执行：

./bin/flink run examples/streaming/WordCount.jar --input ${your_source_file} --output ${your_sink_file}

停止集群：

./bin/stop-cluster.sh

## 常用配置介绍 ##

conf / slaves  
conf / slaves 用于配置 TaskManager 的部署，默认配置下只会启动一个 TaskManager 进程，如果想增加一个 TaskManager 进程的，只需要文件中追加一行“localhost”。  
也可以直接通过“ ./bin/taskmanager.sh start ”这个命令来追加一个新的 TaskManager：  
./bin/taskmanager.sh start|start-foreground|stop|stop-all  
conf/flink-conf.yaml  
conf/flink-conf.yaml 用于配置 JM 和 TM 的运行参数，常用配置有：

# The total process memory size for the JobManager.
    #
    # Note this accounts for all memory usage within the JobManager process, including JVM metaspace and other overhead.
    # 包括JobManager进程中的所有内存使用，包括JVM元空间和其他开销
    jobmanager.memory.process.size: 1600m
    
    
    # The total process memory size for the TaskManager.
    #
    # Note this accounts for all memory usage within the TaskManager process, including JVM metaspace and other overhead.
    # 这包括TaskManager进程中的所有内存使用，包括JVM元空间和其他开销
    taskmanager.memory.process.size: 1728m
    
    # To exclude JVM metaspace and overhead, please, use total Flink memory size instead of 'taskmanager.memory.process.size'.
    # It is not recommended to set both 'taskmanager.memory.process.size' and Flink memory.
    # 要排除JVM元空间和开销，请使用总Flink内存大小，而不是'taskmanager.memory.process.size'。不建议同时设置'taskmanager.memory.process. '和Flink内存
    # taskmanager.memory.flink.size: 1280m
    
    # The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.
    # 每个任务管理器提供的任务槽数。每个插槽运行一个并行管道
    taskmanager.numberOfTaskSlots: 1
    
    # The parallelism used for programs that did not specify and other parallelism.
    # 用于未指定程序的并行度和其他并行度
    parallelism.default: 1

## 日志的查看和配置 ##

JobManager 和 TaskManager 的启动日志可以在 Flink binary 目录下的 Log 子目录中找到。Log 目录中以“flink-\{id\}-$\{hostname\}”为前缀的文件对应的是 JobManager 的输出，其中有三个文件：

*  `flink-${user}-standalonesession-${id}-${hostname}.log`：代码中的日志输出
 *  `flink-${user}-standalonesession-${id}-${hostname}.out`：进程执行时的stdout输出
 *  `flink-${user}-standalonesession-${id}-${hostname}-gc.log`：JVM的GC的日志

Log 目录中以“flink-\{id\}-$\{hostname\}”为前缀的文件对应的是 TaskManager 的输出，也包括三个文件，和 JobManager 的输出一致。

日志的配置文件在 Flink binary 目录的 conf 子目录下，其中：

*  `log4j-cli.properties`：用 Flink 命令行时用的 log 配置，比如执行“ flink run”命令
 *  `log4j-yarn-session.properties`：用 yarn-session.sh 启动时命令行执行时用的 log 配置
 *  `log4j.properties`：无论是 Standalone 还是 Yarn 模式，JobManager 和 TaskManager 上用的 log 配置都是 log4j.properties

这三个“log4j.\*properties”文件分别有三个“logback.\*xml”文件与之对应，如果想使用 Logback 的同学，只需要把与之对应的“log4j.\*properties”文件删掉即可，对应关系如下：  
log4j-cli.properties -> logback-console.xml

log4j-yarn-session.properties -> logback-yarn.xml

log4j.properties -> logback.xml

需要注意的是，“flink-\{id\}-和\{user\}-taskexecutor-\{hostname\}”都带有“，\{id\}”表示本进程在本机上该角色（JobManager 或 TaskManager）的所有进程中的启动顺序，默认从 0 开始。

## 进一步探索 ##

尝试重复执行“./bin/start-cluster.sh”命令，然后看看 Web 页面（或者执行jps命令），看看会发生什么？可以尝试看看启动脚本，分析一下原因。接着可以重复执行“./bin/stop-cluster.sh”，每次执行完后，看看会发生什么。

[fuyun@bigdata-training tools]$ /opt/modules/flink-1.12.0/bin/start-cluster.sh 
    Starting cluster.
    Starting standalonesession daemon on host bigdata-training.fuyun.com.
    Starting taskexecutor daemon on host bigdata-training.fuyun.com.
    [fuyun@bigdata-training tools]$ 
    [fuyun@bigdata-training tools]$ 
    [fuyun@bigdata-training tools]$ /opt/modules/flink-1.12.0/bin/start-cluster.sh 
    Starting cluster.
    [INFO] 1 instance(s) of standalonesession are already running on bigdata-training.fuyun.com.
    Starting standalonesession daemon on host bigdata-training.fuyun.com.
    [INFO] 1 instance(s) of taskexecutor are already running on bigdata-training.fuyun.com.
    Starting taskexecutor daemon on host bigdata-training.fuyun.com.
    [fuyun@bigdata-training tools]$ 
    [fuyun@bigdata-training tools]$ 
    [fuyun@bigdata-training tools]$ /opt/modules/flink-1.12.0/bin/start-cluster.sh 
    Starting cluster.
    [INFO] 1 instance(s) of standalonesession are already running on bigdata-training.fuyun.com.
    Starting standalonesession daemon on host bigdata-training.fuyun.com.
    [INFO] 2 instance(s) of taskexecutor are already running on bigdata-training.fuyun.com.
    Starting taskexecutor daemon on host bigdata-training.fuyun.com.
    [fuyun@bigdata-training tools]$ 
    [fuyun@bigdata-training tools]$ /opt/modules/flink-1.12.0/bin/stop-cluster.sh 
    Stopping taskexecutor daemon (pid: 15368) on host bigdata-training.fuyun.com.
    No standalonesession daemon (pid: 15053) is running anymore on bigdata-training.fuyun.com.
    [fuyun@bigdata-training tools]$ 
    [fuyun@bigdata-training tools]$ /opt/modules/flink-1.12.0/bin/stop-cluster.sh 
    Stopping taskexecutor daemon (pid: 14508) on host bigdata-training.fuyun.com.
    No standalonesession daemon (pid: 14132) is running anymore on bigdata-training.fuyun.com.

重复启动会提示已经有多少个standalonesession和taskexecutor实例已经运行了，每次启动taskexecutor会新增一个，但standalonesession不会新增，一直只有一个实例  
通过jps会发现会新增TaskManagerRunner进程，但StandaloneSessionClusterEntrypoint一直只有一个进程  
在重复停止会提示停止的是哪个taskexecute的pid和standalonesession的pid

[fuyun@bigdata-training flink-1.12.0]$ jps
    13284 StandaloneSessionClusterEntrypoint
    39348 DataNode
    38374 NodeManager
    6007 JobHistoryServer
    39289 NameNode
    13595 TaskManagerRunner
    14508 TaskManagerRunner
    16364 QuorumPeerMain
    2894 RunJar
    14671 Jps
    38319 ResourceManager
    [fuyun@bigdata-training flink-1.12.0]$ jps
    13284 StandaloneSessionClusterEntrypoint
    39348 DataNode
    38374 NodeManager
    6007 JobHistoryServer
    15368 TaskManagerRunner
    39289 NameNode
    13595 TaskManagerRunner
    14508 TaskManagerRunner
    16364 QuorumPeerMain
    2894 RunJar
    15439 Jps
    38319 ResourceManager

在观察log目录下的文件情况，每启动一次，standalonesession和taskexecutor后的id会递增：

[fuyun@bigdata-training flink-1.12.0]$ ls log/
    flink-fuyun-standalonesession-0-bigdata-training.fuyun.com.log  flink-fuyun-standalonesession-2-bigdata-training.fuyun.com.log  flink-fuyun-taskexecutor-1-bigdata-training.fuyun.com.log
    flink-fuyun-standalonesession-0-bigdata-training.fuyun.com.out  flink-fuyun-standalonesession-2-bigdata-training.fuyun.com.out  flink-fuyun-taskexecutor-1-bigdata-training.fuyun.com.out
    flink-fuyun-standalonesession-1-bigdata-training.fuyun.com.log  flink-fuyun-taskexecutor-0-bigdata-training.fuyun.com.log       flink-fuyun-taskexecutor-2-bigdata-training.fuyun.com.log
    flink-fuyun-standalonesession-1-bigdata-training.fuyun.com.out  flink-fuyun-taskexecutor-0-bigdata-training.fuyun.com.out       flink-fuyun-taskexecutor-2-bigdata-training.fuyun.com.out

重复启动-停止再启动-停止操作，会发现在standalonesession和taskexecutor的log后会新增`.id`后缀，id会根据你重复操作次数递增

[fuyun@bigdata-training flink-1.12.0]$ ls log/
    flink-fuyun-standalonesession-0-bigdata-training.fuyun.com.log    flink-fuyun-standalonesession-0-bigdata-training.fuyun.com.out  flink-fuyun-taskexecutor-0-bigdata-training.fuyun.com.log.1
    flink-fuyun-standalonesession-0-bigdata-training.fuyun.com.log.1  flink-fuyun-taskexecutor-0-bigdata-training.fuyun.com.log       flink-fuyun-taskexecutor-0-bigdata-training.fuyun.com.out

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70]: /images/20210923/10a96022eced4064a1bee224f971c9e7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 1]: /images/20210923/abd85332ea1c422ea2f33527e7b1c7ce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 2]: /images/20210923/084da56c01b3408396d2b74140d453a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NjM2Mw_size_16_color_FFFFFF_t_70 3]: /images/20210923/12c161aeecc844b2a5fcbc9982bb40e6.png