Flink简介

Flink 起源于 Stratosphere 项目，Stratosphere 是在 2010~2014 年由 3 所地处柏林的大学和欧洲的一些其他的大学共同进行的研究项目， 2014 年 4 月 Stratosphere 的代码被复制并捐赠给了 Apache 软件基金会， 参加这个孵化项目的初始成员是 Stratosphere 系统的核心开发人员，2014 年 12 月， Flink 一跃成为 Apache 软件基金会的顶级项目。

在德语中， Flink 一词表示快速和灵巧， 项目采用一只松鼠的彩色图案作为logo这不仅是因为松鼠具有快速和灵巧的特点， 还因为柏林的松鼠有一种迷人的红棕色， 而 Flink 的松鼠 logo 拥有可爱的尾巴， 尾巴的颜色与 Apache 软件基金会的 logo 颜色相呼应， 也就是说， 这是一只 Apache 风格的松鼠。

简单来说，Flink 是一个分布式的流处理框架，它能够对有界和无界的数据流进行高效的处理。Flink 的核心是流处理，当然它也能支持批处理，Flink 将批处理看成是流处理的一种特殊情况，即数据流是有明确界限的。这和 Spark Streaming 的思想是完全相反的，Spark Streaming 的核心是批处理，它将流处理看成是批处理的一种特殊情况， 即把数据流进行极小粒度的拆分，拆分为多个微批处理。

Flink集群部署

Flink支持多种安装模式：

• Local - 本地单机模式，用于学习测试
• Standalone——独立集群模式，Flink自带集群，开发测试环境使用
• StandaloneHA - StandaloneHA - StandaloneCluster 高可用模式，Flink自带集群，开发测试环境使用
• Flink On Yarn/Mesos/K8s… - 计算资源由 Hadoop YARN 管理并用于生产环境

下面部署StandaloneHA ，准备3台虚拟机，节点规划：

主机名	IP地址	角色
bigdata1	192.168.92.10	jobmanager
bigdata2	192.168.92.11	jpbmanager taskmanager
bigdata3	192.168.92.12	taskmanager

Standalone 集群实际上只有一个 JobManager，此时是存在单点故障的，所以官方提供了 Standalone Cluster HA 模式来实现集群高可用。

在 Standalone Cluster HA 模式下，集群可以由多个 JobManager，但只有一个处于 active 状态，其余的则处于备用状态，Flink 使用 ZooKeeper 来选举出 Active JobManager，并依赖其来提供一致性协调服务，所以需要预先安装 ZooKeeper 。

两个JobManager,当主 JobManager 宕机之后，使用备用 JobManager ,等宕机的 JobManager 恢复之后，又变成备用。

另外在高可用模式下，还需要使用分布式文件系统来持久化存储 JobManager 的元数据，最常用的就是 HDFS，所以 Hadoop 也需要预先安装。

关于 Hadoop 集群和 ZooKeeper 集群的搭建可以参考：

• Hadoop 集群搭建
• Zookeeper 集群搭建

flink下载地址：https://archive.apache.org/dist

flink安装参考：https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/deployment/ha/overview/

以下操作在所有节点执行。

配置主机名解析

cat > /etc/hosts <<EOF
192.168.92.10 bigdata1
192.168.92.11 bigdata2
192.168.92.12 bigdata3
EOF

下载Flink安装包

wget https://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.13.1/flink-1.13.1-bin-scala_2.12.tgz

解压安装

tar -zxvf flink-1.13.1-bin-scala_2.12.tgz -C /opt
mv /opt/flink-1.13.1/ /opt/flink

配置环境变量

cat > /etc/profile.d/flink.sh <<'EOF'
export FLINK_HOME=/opt/flink
export PATH=$FLINK_HOME/bin:$PATH
EOF

source /etc/profile

确认安装成功

flink --version

修改 conf/flink-conf.yaml 文件，增加如下配置：

cat > /opt/flink/conf/flink-conf.yaml <<EOF
#basic
jobmanager.rpc.address: localhost
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 1600m
taskmanager.memory.process.size: 1728m
taskmanager.numberOfTaskSlots: 2
parallelism.default: 6
jobmanager.execution.failover-strategy: region
io.tmp.dirs: /opt/flink/tmp

high-availability: zookeeper
high-availability.zookeeper.quorum: bigdata1:2181,bigdata2:2181,bigdata3:2181
high-availability.zookeeper.path.root: /flink
high-availability.cluster-id: /cluster_one
high-availability.storageDir: hdfs://bigdata1:9000/flink/ha/
EOF

参数说明：

• high-availability: zookeeper # 配置使用zookeeper来开启高可用模式
• high-availability.zookeeper.quorum: hadoop003:2181 # 配置zookeeper的地址，采用zookeeper集群时，可以使用逗号来分隔多个节点地址
• high-availability.zookeeper.path.root: /flink # 在zookeeper上存储flink集群元信息的路径
• high-availability.cluster-id: /standalone_cluster_one # 集群id
• high-availability.storageDir: hdfs://hadoop001:8020/flink/recovery # 持久化存储JobManager元数据的地址，zookeeper上存储的只是指向该元数据的指针信息

修改masters配置master节点

cat > /opt/flink/conf/masters <<EOF
bigdata1:8081
bigdata2:8081
EOF

修改workers配置workers节点

cat > /opt/flink/conf/workers<<EOF
bigdata2
bigdata3
EOF

下载hadoop依赖包，复制到所有节点$FLINK_HOME/lib目录下：

下载地址：https://flink.apache.org/downloads.html#additional-components

cp flink-shaded-hadoop-2-uber-*.jar /opt/flink/lib/

确保 Hadoop 和 ZooKeeper 已经启动后，使用以下命令来启动集群：

[root@bigdata1 ~]# start-cluster.sh
Starting HA cluster with 2 masters.
Starting standalonesession daemon on host bigdata1.
Starting standalonesession daemon on host bigdata2.
Starting taskexecutor daemon on host bigdata2.
Starting taskexecutor daemon on host bigdata3.

可以看到集群已经以 HA 的模式启动，此时还需要在各个节点上使用 jps 命令来查看进程是否启动成功，正常情况如下：

[root@bigdata1 ~]# jps
21072 ResourceManager
19762 NameNode
20818 SecondaryNameNode
21411 NodeManager
24044 Jps
19919 DataNode
23679 StandaloneSessionClusterEntrypoint

[root@bigdata2 ~]# jps
13638 StandaloneSessionClusterEntrypoint
13902 TaskManagerRunner
14079 Jps

[root@bigdata3 ~]# jps
10341 TaskManagerRunner
10439 Jps

在bigdata1和 bigdata2中运行 JobManager 进程，在bigdata2和 bigdata3中运行 TaskManager 进程。分别打开访问，两个页面都可以查看集群信息：

http://bigdata1:8081/
http://bigdata2:8081/

我们可以 kill掉bigdata1机器上的Job Manager,然后备用(bigdata2)Job Manager也是可以使用的。

./bin/jobmanager.sh stop

再启动bigdata1的Job Manager

./bin/jobmanager.sh start