开发者学堂课程【Apache Flink 入门到实战 - Flink 开源社区出品 ：Flink 安装部署、环境配置及运行应用程序（二）】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/632/detail/10034
Flink 安装部署、环境配置及运行应用程序（二）
二、运行 Flink
(1)基本概念
运行 Flink 应用其实非常简单,但是在运行 Flink 应用之前,还是有必要了解 Flink 运行时的各个组件,因为这涉及到 Flink 应用的配置问题。
map 是将一个元素转换为另一个元素，keyby 不同元素根据某种方法把它转换到某一个具体的执行中，图如下。Window 把一个窗口内的数据集中起来，接着方法处理，结果发给 sink，写到外部存储。本图上面是逻辑上的图，下面是更近一步具体执行。
过程被切成三段，切分开是因为 map 一个元素转换为另一个元素，可以在一个执行完成，在 subtask，当 map 进去 keyby 之后，不同的 key 进去到不同的执行中，map 结果最后会塞到 subtask。整个过程是，在这里面需要把数据并把处理，因为 keyby，导致中间切了一刀，让这分为两个去处理。
通过这张图我们可以看到,在一个 DAG 图中，不能被 chainoperator 在一起的会被分隔到不同的 Task 中，也就是说，Task 是 Flink 中资源调度的最小单位。
整个 flink 分为三个角色，第一个是 flink program 提交 flink 作业的工具，将用户的代码进行优化最后通过 client 端传给 job manager。它是整个过程中的控制的角色它调度 job，包括统一管理相应的流程。最后干活的人是 task manager，它可以发 task slot 就是用来执行具体的某个 subtask 的容器。比如，放了三个 task slot，sub manager 可以同时启动三个 subtask，它既可以是第一阶段也可以中间阶段，也可以是 sink。
task manager 上面除了是管理 task 具体 subtask 具体执行的容器以外，它维护相应的资源，其中的资源包括 memory，network，actor system，三大块。
task manager 这个角色是在以后的演示过程中会需要着重注意，以后 flink 的部署离不开这两个角色。
Flink运行时包括两类进程
JobManager(又称为 JobMaster ):协调 Tsk 的分布式执行，包括调度 Task、协调创 checkpointjob 以及当 failover 时协调各个Task 从 checkpoint 恢复等。
TaskManager( 又称为 Worker):执行 dataflow 中的，包括内存 buffer 分配、 Data Stream 的传递等。
task manager 上有三个 slot，既可以跑 source，也可以跑 keyback，还有 sink。可以看到 task slot 就是 subtask 的容器。一个 taskmanager，配置了多少 task slot 意味着可以同时跑多少个 subtask，这个概念在之后的启动 flink 集群的时候着重关注。因为 task manager上的 task slot 配置太少，导致任务启动不了。
(2)运行环境准备
1.准备 Flink binary
接下来我们需要准备 Flink binary,主要有两种方法:
a. 直接从 FlinkFlink 官网上下载 binary 的压缩包:
b. https://fink.apache.org/downloads.html
b.从Flink源码编译而来(参考第一章节的第5点编译 flink代码)
下图所示fink-dist/target/flink--1.5.1.ar.gz是flink binary的压缩包
Flinkcode 的 pathbuild-target 下的-目录就是一个可用的 fink binary 目录
链接上面可以看到flink提供了1.2/1.5/1.6三个版本，每个版本里面又出了几个子版本。比如第一个说明没有很多依赖，只有flink核心依赖。这个包里说明通过2.8打包，会把 jar 包 sheed 这个 barrel，不需要格外的依赖，就可以运行。如果没有编译代码，可以直接从这里面下载，选择合适的镜像源，具体去下载。
tgz 这个文件其实和编译出来的文件，下载的这个文件和编译出来的文件等价，既可以用编译出来的包，也可以下载的包去运行。
将下载或编译出来的 Flink binary 的压缩包解压后，和编译出来的 buid- target 目录是等价的。
有 target 解压出来，放在指定的目录下，example 主要是一些可以测试用的 jar包。bin 存放启动各种 flink 脚本。library 保存 flink 主要的 jar 包，如果用Hadoop，这个包是必须的。
opt 不会默认带上，如果需要这个包，可以直接把它拷贝到 flink 目录下。
conf 是本次一开始启动运行需要目录，最核心就是 flink conf yaml。
tar zxvf xxx. tar.gz -C $(dir_ for_decompression)
2.安装 Java，并配置 JAVA HOME 环境变量
第一章中主要介绍的是 Flink 开发环境的部署，而如果只是需要运行 Flink binary，则只需要安装和配置好 java 即可。
(3)单机 standaloneflink
1.基本的启动流程
最简单的运行 flink 应用的方法 就是以单机 standalone 的方式运行，进入 flink binary 的目录下(一般情况下不需要修改配置文件)，执行：
bin/start-cluster-.sh
其他不用配置。
输出结果如下，表示启动正常:
启动成功后，打开tt:127.0.0.1:8081/就能看到
http://127.0.0.1:8081/ flinkweb 的 web界面:
直接在本地上看到启动的一个 flink local 集群，这里面有一个 task manager 和四个 task slot。
因为 numberoftasksslots 表示每 个 task manager 有多个 task slot，只启动一个task manager，默认情况下，启动一个。
现在 task manager 有三个，task slot 有12个。
提交一个 Word Count 的任务，命令如下:
./bin/flink run examples/streaming/WordCount.jar
可以在 web 界面上看到 job 完成了。
因为跑的比较快，所以看到的已经结束了。前面是一个 map，先通过 hash，聚合在输出。taskmanager 日志，因为是本地模式，几个日志都在一起。如果以后查问题可以通过这种方式查看日志。
一个目录下，大概都是一样的。这个是最简单的集群。
点击 job 信息，可以看到 word count 执行图。点进最下面的 vertex，可以通过stdout 信息看到 word count 的结果。
看一下结果，在 out 里面。
结合 Flink 代码:
flink-examples->flink-examples-streaming-> wordcount-> class
WordCount
可以看到，不传任何参数时， input的输入是代码中的一个字符串。
接着我们尝试通过 input" 参数指定我们自己的本地文件作为输入，然后执
行:
/bin/flink run examples/streaming/WordCount. jar--input
{your_source_file}
2.常用配置介绍
xm 参数表示初始大小，在运行方面做了更机械化的控制，和文本有一定的区别。
我们在本机上执行 jps 命令,可以看到 Flin 相关的进程主要有两个，一个是JobManager 进程，另一个是 TaskManager 进程。我们可以进一步用 ps 命令看看进程的启动参数:
结合 Flink binary-c 目录下的 conf 子目录中的 fink-yaml 文件，
常用的配置有下面几个：
The heap size for the JobManager JVM
jobmanager. heap.: 1024
The heap size for the TaskManager JVM
taskmanager.heap.mb: 1024
# The number of task slots that each TaskManager offers. Each slot runs
one parallel pipeline.
taskmanager. numberOfTaskSlots: 4
the managed memory size for each task manager.
taskmanager.managed.memory.size: 256
我们可以先用下面这个命令停掉 standalone 集群：
./bin/stop-cluster.sh
然后修改 ink-conf.yaml 中的这几个配置如下:
The heap size for the JobManager JVM
jobmanager.heap.mb: 1024
The heap size for the TaskManager JVM
然后重启 standalone 集群:
/bin/start-cluster.sh I
启动成功后，打开 http: //127.0.0.1:8081/Flinkweb :就能看到 Flink 的 web 界面。
问题：发现启动一个 standalone 集群发现 task manager 变多了。其实原因很简单，再启动一次，还是可以起来，数量有了变化。如果之前shengyang不停，可以通过日志看到，需要绑定一个固定的端口，又启动了三个jps。只需要把这个cluster 停掉，每次停会按照 conf slaves 列的 jps 数量去删减。
另外还需要注意，本机上有两份 flink memory。先起来 1.7flink standalone 集群，再启动1.5，发现界面是1.7的，其实是因为1.5的界面没起来，tm注册到1.7，可能会导致 flink 自动升级一样，其实不是。
一个人配置，另外一个人 log，log 是很关键。log 的格式，第一个是 flink，第二个是用户名，第三个是角色名。
这个角色既可以是 task executor，也可以是 alone session。然后另外 id 表示当前角色的 id，以后是 host。
id 是自增的，再启动一次看看。启动三个 tmp，一般是在 tmp 目录下写一个 tmd 文件。flink executor 的 pid 对应三个 task manager。它把每个角色启动的 pid 存放 pid 文件中，之后 pid 可以用来表示文件，用于 stop 进程 cue 哪 一个进程。stand alone 通过 pid 文件去管理每个角色，这个细节了解一下。
pid 在不同目录下执行的都会自增，因为使用公共的路径。
我们再看一下启动参数
在 blink 开源分支上,，TaskManager 的内存计算上相对于现在的社区版本要更精细化， TaskManager 进程的堆内存限制(-xmx)一般的计算方法是:
TotalHeapMemory taskmanager heapmb+
taskmanager. managed. memory.size+
taskmanagerprocess.heap. memorymb(默认值为128MB)
相对地，最新的 fink 社区 release-版本-1.7中 JobManage 和的默认内存配置改为:
The heap size for the JobManager JVM
jobmanager.heap.size: 1024m
The heap size for the TaskManager JVM
taskmanagerheap.size:1024m
zoo.cfg 是 flink 自带的启动 zookeeper suber 脚本的配置，可以启动 zookeeper集群。master 和 slaves 都是 standalone下使用。下面有log4 文件。如果两个文件同时存在，把两个文件放在环境分压中。默认情况下，两个都放，只有 log4j 会生效。这三个是互相配对的。如果想用 logback 作为 log，把 log4j 删掉。
其中:
log4j-cli. properties:用 Flink 命令行时用的 log 配置,比如执行 flink run
命令
log4j-yarm- -session. properties:是用 yar session.sh启动时命令行执行
时用的 log 配置
log4j .properties:无论是 standalone 还是 an 模式, JobManager 和
需要注意的是，
fink-suser--standalonesession--sid-Shostname-和flink-Suser-taskexecutor---sid-shostname)都带有“id,sid 表示本
进程在本机上该角色(《JobManagerTaskManagerj《或 TasManager)的所有进程中的启动顺序，默认从0开始。
3.进一步探索
接下来我们可以在上面的基础上再次调用这个命令(重复调用 start-cluster-.sh 命令)
/bin/start-cluster.sh
命令的输出结果如下所示
l tarting toskexecutor doemon on host -6c96cfd97dcb. local.
再打开 http://127.0.0.1:8081/Flinkweb:就能看到Flink的web界面:
可以看到，每个pid文件中最末尾的pid被取出来并被kill了
(4)多机部署 Flink standalone 集群
部署前要注意的要点:
1.每台机器上配置好 java 以 JAVA 及 HOME 环境变量
2.最好挑选一台机器,和其他机器 ssh 打通
3.每台机器上部署的 Flink binary 的目录要保证是同一个目录
4.如果需要用 hdfs,需要配置 HADOOP CONF DIR 环境变量配置上
JobManager 机器:z05f06378.sa.zth.tbsite.net
TaskManager 机器:z05f06378.sqa.zth. tbsite.net
z05c19426.sqa. zth. tbsite. netz05f10219sqa.zth.tbsite.net
这边还有其他几个机器，使用命令把压缩包拷贝到其他机器的同一个目录下去，这个 pgmscp 命令其实是阿里内部的批量传输文件的命令，等价于 pscp。
jobmanager.rpc. address: z05f06378.sqa. zth. tbsite. net
再执行命令，把它挤压出来。
现在另一个机器上也有了，现在已经把 flink 的部署在了目录下，把修改过的三个目录同步到机器上。现在把所有机器同步完成，现在执行 start -cluster。
(5) standaloneHigh 模式的 Availability(HA) 部署和配置
由下面这幅架构图，我们可以看到 JobManager 是整个系统中最可能导致系统不可用的角色。一个 TaskManager 了，如果资源足够(空闲 TaskSlot 足够)的话,则只需要把相关 task 调度到其 Ta 他空闲上,然后 job 从 checkpoint 中恢复即可。而如果当前集群中只配置了一个 obManager，则一旦 JobManager 挂了，就必须等待这个 JobManager重新恢复,如果恢复时间过长，就可能导致整 个job 失败。
因此如果在生产业务使用 standalone 模式,则需要部署配置 HighAvailability，这样同时可以有多个 JobManager 待命,从而使得 JobManager 能够持续服务。
如果想尝试试用 Flink standalone HA 模式需要确保基于 flink release--1.6.1 及以上版本，因为这里社区有个bug会导致这个模式下主 JobManager 不能正常工作。因 fink 中 blinkflink 的开源分支是基于 release--1.5.1，所以也存在这个问题。问题的详情见:
https://stackoverflow.com/questions/53869850/fliailed
可以直接从 FlinkFlink 官网上下载最新版本的 binary 的压缩包试用：
https://ink.apache.org/downloads. html ,.org/downloads.html，下载时候需要注意，接下来的实验中需要用到hdfs,所以需要下载带有 hadoopflin支持的 binary包否则接下来初始化 standalone 集群的时候会报错，初始化 hdfs filesystem 失败。
1.(可选)使用 fink 自带的脚本部署 zk
Fink 现在的 HA 需要 zookeeper 基于,如果你的集群中没有，Flink提供了启动zookeeper 集群的脚本。首先修改配置文件 confzoo.cfg, 根据你要部署的 zookeeper server 的机器数来配置 erver.x=addressX peerPort leaderPort，其中x是一个 ookeeper server 的唯一 D,且必须是数字。
PS:端口一般不需要配置,因为我的环境上已经部署了一套 zookeeper，所以会导致端口占用而启动不起来，所以改了默认端口。)
然后后动 zookeeper:
/bin/start-zookeeper-quorumIsh
jps 命令看到 zk 进程已经启动:
Sips
20348ps
220415 FlinkZooKeeperQuorumPeer
停掉 zookeeper 集群
/bin/stop-zookeeper-quorum. sh
沙晟阳
(6)修改 flink standalone 集群的配置
停掉之前启动到 standalone 集群
/bin/stop-cluster.sh
修改 confmasters文件,增加一个 JobManager
Scat conf/masters
z05f06378.sqa.zth. tbsite. net: 8081
z05c19426.sqa. zth. bsite. net: 8081
之前修改过的 conf/slaves 文件保持不变:
Scat conf/slaves
z05f06378.sqa.zth. tbsite. net
z05c19426..zth.tbsite.net
z05f10219.sqa.zth.tbsite.net
修改 conffink-conyaml 文件:
#配置 high-availability- mode
high-availability:zookeeper
#配置 zookeeper quorum(hostname 和端口需要依据对应zk的实际配置)
high-availability, zookeeper, qucrurn:
z05f02321.sqa.zth. tbsite.net:218120510215.sga.zth.tbsite.net:2181
#(可选}设置 zookeeper 的 root 目录
high-avallability.zookeeper.path.root: Mest_din'test_standalone2_root
#(可选)相当于是这个 standalonozk 集群中创建的 nodeejnamespace
high-availability.cluster-id: Mest_dir/test_standalone2
# JabManager 的 meta 信息放在 dfs.在 zk 上主要会保存一个指向 dfs 路径的指针
high-availability. storageDir: hdfs:est_/recovery2/:
如果想尝试试用 Flink standalone HA 模式,需要确保基于 flink release--16.1及以上版本，因为这里社区有个 bug 会导致这个模式下主不能正常工作。因为 flink中的 blink 开源分支是基于 release-1.5.1，所以也存在这个问题。问题的详情见：
https://stackoverflow.com/questions/5386985/flink-ha-standalone-cluster-failed
可以直接从 FlinkFlink 官网上下载最新版本的 binary的压缩包试用:
httsi https://flink. apache. org/downloads. html,.org/downloads.html, 下载时候需要注意,接下来的实验中需要用到hdfs，所以需要下载带有 hadoopflink 支持的包否则接下来初始化 standalone 集群的时候会报错，初始化 hdfs filesystem 失 败。
(7)使用 yamn 模式跑 flink job
什么情况下适合使用 yarn 模式 flink 跑 job?
相对于 standalone模 式 yarn 模式 flink 允许 job 的好处有：
1.资源按需使用,提高集群的资源利用率
2.任务有优先级,根据优先级运行作业
3.基于 YARN 调度系统,能够自动化地处理各个角色的 failover
JobManagerTaskManagerYarn 进程和进程都由 NodeManager 监控
如果 JobManage r进程异常退出,则 Yarn ResourceManager 会重新调度 JobManager 到其他机器
如果 TaskManager 进程异常退出, JobManager 会收到消息并重新向 Yarn ResourceManager 申请资源，重新启动 TaskManager
(8).在 YARN 上启 long 动 running 的 fink 集群(yarn session)
我们演示用的 YARN 集群 yarn: http://z05c19394.sqa.zth.tbs net:8088/cluster
查看命令参数:
/bin/yam-session.sh-h
创建一个 YARN 模式 flink 的集群
/bin/yam-session.sh-n 4-im 1024m-tm 4096m
其中用到的参数是:
n.-container <arg> Number of TaskManagers
-im, -jobManagerMemory carg Memory for JobManager Container with optional unit(default: MB)
.-tm, -taskManagerMemory <arg> Memory per TaskManager
Container with optional unit(default: MB)
-tm,-taskManagerMemory <arg> Memory per TaskManager
Container with optional unit(default: MB)
-qu,-queue <arg> Specify YARN queue.
-S.-slots <arg> Number of slots per TaskManager
-t,-ship <arg> Ship files in the specified directory (t for transfer
提交一个 flink job 到 fink 集群:
这次提交 flink job,虽然没有指定对应 yam application 的信息,确可以提交到对应的 flink 集群，原因在于tmp/.yarn--properties-S-user 文件中保存了上次创建 yarn session 的集群信息。所以如果同一用户在同一机器上再次创建个 session，则这个文件会被覆盖掉。
那如果删掉 mpyar -properties---$user 或者在另一个机器上提交作业能否提交到预期到 yar session 中呢?
这也是可以的，如果配置了 HighAvailability，则可以根据 cluster--id，从zookeeperJobManager 上获取到的地址和端口，从而提交作业。
high-availability. cluster-id
如果 Yarn session没有配置HA，又该如何提交呢?
这个时候就必须要在提交 flink job 的命令中指明 YARN 上的 application id，通过-yid 参数传入
我们可以发现，每次跑完任务不久, TaskManager 就没有了，下次在提交任务的时候，TaskManager 又会重新拉起来。如果希望 TaskManager 启动后就持续运行，可以在 conf/ink-conf.yam 文件中置下面这个参数单位是 milliseconds,默认值是30000L，即30秒:
resourcemanager.taskmanager-timeout
(9)在 YARN 上运 flink 行单个 job
如果你只想运行单个 link job 后就退出,那么可以用下面这个命令:
/bin/flink run -m yarn-cluster-yn 2 examples/streaming/WordCount jar
-input hdfs:///est dir/input dir/story--output
hdfsest dirloutput dir/output
常用的配置有:
.-yn,-yarncontainer <arg> Number of Task Managers
.-yqu,--yamqueue carg
Specify YARN queus.
-ys, -yamslots carg>
Number of slots per TaskManager
.-yqu, -yarnqueue carg
Specify YARN queue.
可以通过 help 命令查看 run 的可用参数:
./bin/fink run-h
我们可以看到,“bin/flink run-h 看到的 Options for yarn-cluster- mode 中的 y 和 yarn 为前的参数其实 和 bin/yarn-session.sh-h 命令是一一对应的，语义上也基本一致。
关于-n(在 yarn session 模式下)、-yn"在(yarn single job 模式下)与 p 参数的关系:
a.-n 和 -yn 在社区版本中(release--1.~ release-1.7 中没有实际的控制作用，实际的资源是根据p”参数来申请的，并且 TM 使用完后就会归还
b.在 blink的开源版本中，-n(在yam session 模式下)的作用就是一开始启动指定数量的 TaskManager，之后即使 job 需要更多的 slot,也不会申请新的 TaskManager
c.在 blink 的开源版本中,，yarn single job 模式 yn”表示的是初始 TaskManager的数量,不设置 TaskManager 的上限。(需要特别注意的是，只有加上 yd 参数才能用 single 模式(例如:命 bin/ flink run- yd -m yarn -cluster xxx)
(10)yarn 模式下的 HighAvailability 配置
首先要确保启动 Yam 集群用的 yam-site.xl 文件中的这个配置，这个是 YARN 集群级别 AM 重启的上限。
然后在 conf/ fflink-conf.yaml 文件中配 flink 置这个 jobJobManager 的能够重启的次数(1+9 retries)
yarn.application-attempts: 10
最后再在conf/ink-conf.yam文件中配置上k相关配置，这几个配置的配置方法和 standalone的HA配置方法基本一致，如下所示。
需要特别注意的是:“high-availability. cluster-i-id 这个配置要去掉，因为在Yarn(以及 mesos)模式下,，cluster--id 如果不配置的话，会配置成 Yam 上的application id，从而可以保证唯一性。否则,在提交作业的时候需要用户去。
保证 cluster-id 的全局唯一性。
#配置 high -availability- mode
high-availability: zookeeper
#配置 zookeeper quorum(hostname 和端口需要依据对应 zk 的实际配置)
high-availability. zookeeper. quorum:
z05f02321.sqa. zth.tbsite. net:2181, z05f10215.sqa.zth.tbsite.net:2181
