10.Flink集群安装操作

cd /export/software/

rz 上传

解压:

tar -zxf flink-1.14.0-bin-scala_2.12.tgz -C /export/server/

为flink创建软连接

cd /export/server/

ln -s flink-1.14.0/ flink

cd /export/server/flink

bin/start-cluster.sh

启动后, 通过JPS查看进程信息:

http://node1:8081

cd /export/server/flink

bin/flink run /export/server/flink/examples/batch/WordCount.jar

JobManager 和 TaskManager 的启动日志可以在 Flink binary 目录下的 log 子目录中找到

cd /export/server/flink/log

cd /export/server/flink/conf

vim flink-conf.yaml

修改内容内容:

# jobManager 的IP地址

jobmanager.rpc.address: node1.itcast.cn

# JobManager 的端口号

jobmanager.rpc.port: 6123

# JobManager JVM heap 内存大小

jobmanager.memory.process.size: 1024m 

# TaskManager JVM heap 内存大小

taskmanager.memory.process.size: 1024m 

# 每个 TaskManager 提供的任务 slots 数量大小

taskmanager.numberOfTaskSlots: 2

# 程序默认并行计算的个数

parallelism.default: 1

cd /export/server/flink/conf

vim workers

添加以下内容:

node1.itcast.cn

node2.itcast.cn

node3.itcast.cn

vim /etc/profile

添加:

export HADOOP_CONF_DIR=/export/server/hadoop/etc/hadoop

重新加载环境变量:

source /etc/profile

5) 将资料中的相关jar包全部上传至flink的lib目录下

cd /export/server/flink/lib

rz上传

cd /export/server

scp -r flink-1.14.0/ node2:$PWD

scp -r flink-1.14.0/ node3:$PWD

在node2和node3创建软链接

cd /export/server

ln -s flink-1.14.0/ flink

修改环境变量:

vim /etc/profile

添加内容:

export HADOOP_CONF_DIR=/export/server/hadoop/etc/hadoop

加载环境变量:

source /etc/profile

cd /export/server/flink

bin/start-cluster.sh

1) 首先启动 hadoop集群

start-all.sh

2) 接着在hdfs上创建目录:

hadoop fs -mkdir -p /test/input

3) 在linux上创建wordcount.txt文件:

cd ~

vim wordcount.txt

内容如下:

hadoop hello world hive

hive hadoop hello hadoop

hadoop hive hive

4) 将其上传至hdfs上:

hadoop fs -put /root/wordcount.txt /test/input

5) 执行flink程序:

cd /export/server/flink

bin/flink run /export/server/flink/examples/batch/WordCount.jar --input hdfs://node1:8020/test/input/wordcount.txt

6) 查看web浏览器:

http://node1:8081

操作步骤

说明

1

下载hadoop的组件Jar包

https://mvnrepository.com/

2

将下载的jar文件拷贝到flink安装目录lib目录下（每个节点都需要拷贝）

3

拷贝完成jar到每个节点以后需要重启flink集群

cd /export/server/flink/conf/

vim flink-conf.yaml

修改以下内容:

#开启HA，使用文件系统作为快照存储:126行

state.backend: filesystem

#默认为none，用于指定checkpoint的data files和meta data存储的目录:131行

state.checkpoints.dir: hdfs://node1.itcast.cn:8020/flink-checkpoints

#默认为none，用于指定savepoints的默认目录:135行

state.savepoints.dir: hdfs://node1.itcast.cn:8020/flink-savepoints

#使用zookeeper搭建高可用:81行

high-availability: zookeeper

# 存储JobManager的元数据到HDFS,用来恢复JobManager 所需的所有元数据 :90行

high-availability.storageDir: hdfs://node1.itcast.cn:8020/flink/ha/

high-availability.zookeeper.quorum: node1.itcast.cn:2181,node2.itcast.cn:2181,node3.itcast.cn:2181

cd /export/server/flink/conf

scp flink-conf.yaml node2:$PWD

scp flink-conf.yaml node3:$PWD

cd /export/server/flink/conf

vim flink-conf.yaml

修改:

jobmanager.rpc.address: node2.itcast.cn

cd /export/server/flink/conf

vim masters

修改为以下内容:

node1.itcast.cn:8081

node2.itcast.cn:8081

cd /export/server/flink/conf

scp masters node2:$PWD

scp masters node3:$PWD

1) 先启动zookeeper集群: 三个节点启动

cd /export/server/zookeeper/bin/

./zkServer.sh start

2) 启动hadoop集群

start-all.sh

3) 启动flink集群

cd /export/server/flink

./bin/start-cluster.sh

操作步骤

说明

1

打开yarn配置页面（每台hadoop节点都需要修改）

vim etc/hadoop/yarn-site.xml

添加

<property>

    <name>yarn.nodemanager.vmem-check-enabled</name>

    <value>false</value>

</property>

是否启动一个线程检查每个任务正使用的虚拟内存量，如果任务超出分配值，则直接将其杀掉，默认是true。

在这里面我们需要关闭，因为对于flink使用yarn模式下，很容易内存超标，这个时候yarn会自动杀掉job

2

分发yarn-site.xml到其它服务器节点

scp yarn-site.xml node2:$PWD

scp yarn-site.xml node3:$PWD

3

启动HDFS、YARN集群

start-all.sh

操作步骤

说明

1

yarn-session.sh(开辟资源)+flink run(提交任务)

这种模式下会启动yarn session，并且会启动Flink的两个必要服务：JobManager和Task-managers，然后你可以向集群提交作业。同一个Session中可以提交多个Flink作业。需要注意的是，这种模式下Hadoop的版本至少是2.2，而且必须安装了HDFS（因为启动YARN session的时候会向HDFS上提交相关的jar文件和配置文件）

通过./bin/yarn-session.sh脚本启动YARN Session

脚本可以携带的参数：

-n(--container)：TaskManager的数量。(1.10 已经废弃)

-s(--slots)： 每个TaskManager的slot数量，默认一个slot一个core，默认每个taskmanager的slot的个数为1，有时可以多一些taskmanager，做冗余。

-jm：JobManager的内存（单位MB)。

-q：显示可用的YARN资源（内存，内核）;

-tm：每个TaskManager容器的内存（默认值：MB）

-nm：yarn 的appName(现在yarn的ui上的名字)。  

-d：后台执行。

注意：

如果不想让Flink YARN客户端始终运行，那么也可以启动分离的 YARN会话。该参数被称为-d或--detached。

确定TaskManager数：

Flink on YARN时，TaskManager的数量就是：max(parallelism) / yarnslots（向上取整）。例如，一个最大并行度为10，每个TaskManager有两个任务槽的作业，就会启动5个TaskManager。

2

启动：

bin/yarn-session.sh -tm 1024  -s 4 -d

上面的命令的意思是，每个 TaskManager 拥有4个 Task Slot（-s 4），并且被创建的每个 TaskManager 所在的YARN Container 申请 1024M  的内存，同时额外申请一个Container用以运行ApplicationMaster以及Job Manager。

TM的数量取决于并行度，如下图：

执行：bin/flink run -p 8 examples/batch/WordCount.jar

3

启动成功之后，控制台显示：

4

去yarn页面：ip:8088可以查看当前提交的flink session

5

然后使用flink提交任务

bin/flink run examples/batch/WordCount.jar

在控制台中可以看到wordCount.jar计算出来的任务结果

6

在yarn-session.sh提交后的任务页面中也可以观察到当前提交的任务：

7

点击查看任务细节：

8

停止当前任务：

yarn application -kill  application_1527077715040_0007

bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar

常用参数：

• -p 程序默认并行度

下面的参数仅可用于 -m yarn-cluster 模式

• -yjm JobManager可用内存，单位兆
• -ynm YARN程序的名称
• -yq 查询YARN可用的资源
• -yqu 指定YARN队列是哪一个
• -ys 每个TM会有多少个Slot
• -ytm 每个TM所在的Container可申请多少内存，单位兆
• -yD 动态指定Flink参数
• -yd 分离模式（后台运行，不指定-yd, 终端会卡在提交的页面上）

在8088页面观察：

yarn application -kill application的ID

注意：

在创建集群的时候，集群的配置参数就写好了，但是往往因为业务需要，要更改一些配置参数，这个时候可以不必因为一个实例的提交而修改conf/flink-conf.yaml;

可以通过：-yD <arg>                        Dynamic properties

来覆盖原有的配置信息：比如：

bin/flink run -m yarn-cluster -yD fs.overwrite-files=true examples/batch/WordCount.jar

-yD fs.overwrite-files=true -yD taskmanager.network.numberOfBuffers=16368

bin/flink run-application -h 

参数：

Options for Generic CLI mode:

     -D <property=value>  Generic configuration options for execution/deployment and for the configured executor.The available options can be found at

                           https://ci.apache.org/projects/flink/flink-docs-stable/ops/config.html

     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is also available with the "Application Mode". The name of the executor to be used for executing the

                           given job, which is equivalent to the "execution.target" config option. The currently available executors are: "collection", "remote",

                           "local", "kubernetes-session", "yarn-per-job", "yarn-session".

     -t,--target <arg>     The deployment target for the given application, which is equivalent to the "execution.target" config option. The currently available targets are:

                           "collection", "remote", "local", "kubernetes-session", "yarn-per-job", "yarn-session", "yarn-application" and "kubernetes-application".

第一种方式

带有 JM 和 TM 内存设置的命令提交：

./bin/flink run-application -t yarn-application \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dyarn.application.name="MyFlinkWordCount" \

./examples/batch/WordCount.jar --output hdfs://node1:8020/wordcount/output_51

第二种方式

在上面例子 的基础上自己设置 TaskManager slots 个数为3，以及指定并发数为3：

./bin/flink run-application -t yarn-application -p 3 \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dyarn.application.name="MyFlinkWordCount" \

-Dtaskmanager.numberOfTaskSlots=3 \

./examples/batch/WordCount.jar --output hdfs://node1:8020/wordcount/output_52

当然，指定并发还可以使用 -Dparallelism.default=3，而且社区目前倾向使用 -D+通用配置代替客户端命令参数(比如 -p)。所以这样写更符合规范：

./bin/flink run-application -t yarn-application \

-Dparallelism.default=3 \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dyarn.application.name="MyFlinkWordCount" \

-Dtaskmanager.numberOfTaskSlots=3 \

./examples/batch/WordCount.jar --output hdfs://node1:8020/wordcount/output_53

第三种方式

和 yarn.provided.lib.dirs 参数一起使用，可以充分发挥 application 部署模式的优势：我们看 官方配置文档 对这个配置的解释：

yarn.provided.lib.dirs: A semicolon-separated list of provided lib directories. They should be pre-uploaded and world-readable. Flink will use them to exclude the local Flink jars(e.g. flink-dist, lib/, plugins/)uploading to accelerate the job submission process. Also YARN will cache them on the nodes so that they doesn't need to be downloaded every time for each application. An example could be hdfs://$namenode_address/path/of/flink/lib

意思是我们可以预先上传 flink 客户端依赖包 (flink-dist/lib/plugin) 到远端存储(一般是 hdfs，或者共享存储)，然后通过 yarn.provided.lib.dirs 参数指定这个路径，flink 检测到这个配置时，就会从该地址拉取 flink 运行需要的依赖包，省去了依赖包上传的过程，yarn-cluster/per-job 模式也支持该配置。在之前的版本中，使用 yarn-cluster/per-job 模式，每个作业都会单独上传 flink 依赖包(一般会有 180MB左右)导致 hdfs 资源浪费，而且程序异常退出时，上传的 flink 依赖包往往得不到自动清理。通过指定 yarn.provided.lib.dirs，所有作业都会使用一份远端 flink 依赖包，并且每个 yarn nodemanager 都会缓存一份，提交速度也会大大提升，对于跨机房提交作业会有很大的优化。
使用示例如下：
my-application.jar 是用户 jar 包

上传 Flink 相关 plugins 到hdfs

cd /export/server/flink/plugins

hdfs dfs -mkdir /flink/plugins

hdfs dfs -put \

external-resource-gpu/flink-external-resource-gpu-1.14.0.jar \

metrics-datadog/flink-metrics-datadog-1.14.0.jar \

metrics-graphite/flink-metrics-graphite-1.14.0.jar \

metrics-influx/flink-metrics-influxdb-1.14.0.jar \

metrics-jmx/flink-metrics-jmx-1.14.0.jar \

metrics-prometheus/flink-metrics-prometheus-1.14.0.jar \

metrics-slf4j/flink-metrics-slf4j-1.14.0.jar \

metrics-statsd/flink-metrics-statsd-1.14.0.jar \

/flink/plugins

根据自己业务需求上传相关的 jar

cd /export/server/flink/lib

hdfs dfs -mkdir /flink/lib

hdfs dfs -put flink-csv-1.14.0.jar \

flink-dist_2.12-1.14.0.jar \

flink-json-1.14.0.jar \

flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-173-9.0.jar \

commons-cli-1.4.jar \

flink-shaded-zookeeper-3.4.14.jar \

flink-table_2.12-1.14.0.jar \

log4j-1.2-api-2.14.1.jar \

log4j-api-2.14.1.jar \

log4j-core-2.14.1.jar \

log4j-slf4j-impl-2.14.1.jar \

/flink/lib

上传用户 jar 到 hdfs

cd /export/server/flink

hdfs dfs -mkdir /flink/user-libs

hdfs dfs -put ./examples/batch/WordCount.jar /flink/user-libs

提交任务

bin/flink run-application -t yarn-application \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dtaskmanager.numberOfTaskSlots=2 \

-Dparallelism.default=2 \

-Dyarn.provided.lib.dirs="hdfs://node1:8020/flink/lib;hdfs://node1:8020/flink/plugins" \

-Dyarn.application.name="batchWordCount" \

hdfs://node1:8020/flink/user-libs/WordCount.jar --output hdfs://node1:8020/wordcount/output_54

也可以将 yarn.provided.lib.dirs 配置到 conf/flink-conf.yaml，这时提交作业就和普通作业没有区别了：

./bin/flink run-application -t yarn-application \

-Djobmanager.memory.process.size=1024m \

-Dtaskmanager.memory.process.size=1024m \

-Dyarn.application.name="MyFlinkWordCount" \

-Dtaskmanager.numberOfTaskSlots=3 \

/local/path/to/my-application.jar

注意：如果自己指定 yarn.provided.lib.dirs，有以下注意事项：

1. 需要将 lib 包和 plugins 包地址用;分开，从上面的例子中也可以看到，将 plugins 包放在 lib 目录下可能会有包冲突错误
2. plugins 包路径地址必须以 plugins 结尾，例如上面例子中的 hdfs://node1:8020/flink/plugins
3. hdfs 路径必须指定 nameservice(或 active namenode 地址)，而不能使用简化方式(例如 hdfs://node1:8020/flink/libs)

该种模式的操作使得 flink 作业提交变得很轻量，因为所需的 Flink jar 包和应用程序 jar 将到指定的远程位置获取，而不是由客户端下载再发送到集群。这也是社区在 flink-1.11 版本引入新的部署模式的意义所在。

Application 模式在停止、取消或查询正在运行的应用程序的状态等方面和 flink-1.11 之前的版本一样，可以采用现有的方法。