【Flink】1.Flink集群部署-Toy模板网

这篇具有很好参考价值的文章主要介绍了【Flink】1.Flink集群部署。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

0 引言

Flink可以部署于各种各样的集群之中，比如Flink自己的standalone集群（不依赖于其他资源调度框架，是Flink自带的），flink on yarn集群等。而不管是standalone还是flink on yarn都属于集群，还有一种特殊的单机flink——local。
Flink真正用来做执行操作的叫做worker，进程在不同的环境模式下运行，名称有所不同。如在Standalone集群模式启动，JobManager叫做StandaloneSessionClusterEntrypoint，TaskManager中叫做TaskManagerRunner，而使用yarn集群启动，名称又不相同。

1. Standalone模式

1.1首先保证当前虚拟机已完成伪集群（1主2从）搭建（默认已进行过SSH免密配置），且均有如下配置：

hadoop-2.7.4
java 1.8
zookeeper 3.4.14
os:centos 7.5

1.2到scala社区，找到合适版本的scala安装包进行下载

wget +网址

1.3到flink社区，找到相应与hadoop版本相吻合的flink解压包，复制网址后到虚拟机上进行安装

wget +网址

1.4解压下载好的压缩包

tar -zxvf flink-1.7.1-bin-hadoop27-scala_2.11.tgz
tar -zxvf /scala-2.11.8

1.5修改环境变量,在profile文件中加入scala和flink的环境配置

vi /etc/profile
flink集群部署,大数据,flink,scala,hadoop

1.6 修改在flink文件夹下的conf文件夹中的flink-conf.yaml配置文件，先配置简单standalone版本的模式

vi …/conf/flink-conf.yaml

jobmanager.rpc.address: 192.168.2.100(主机ip地址）
jobmanager.rpc.port: 6123
jobmanager.heap.size: 1024m
taskmanager.heap.size: 1024m
taskmanager.numberOfTaskSlots: 2
parallelism.default: 1

1.7修改在flink文件夹下的conf文件夹中slaves加入两台从节点的ip地址

vi …/conf/slaves
flink集群部署,大数据,flink,scala,hadoop

1.8 修改在flink文件夹下的conf文件夹中master文件，加入主节点ip地址

vi …/conf/master
localhost:8081
注：此配置关系到在web查看flink运行的整体情况

1.9将在主节点上配置好的flink文件夹统一分发到相应的两台从节点上

Scp -r flink root@slave1.hadoop:/usr/local/flink
Scp -r flink root@slaves2.hadoop:/usr/local/flink

1.10启动flink集群，查看是否可用

…/bin/start-cluster.sh
分别查看主节点jps是否有 StandaloneSessionClusterEntrypoint
flink集群部署,大数据,flink,scala,hadoop

从节点jps是否有TaskManagerRunner
flink集群部署,大数据,flink,scala,hadoop

1.11打开浏览器输入local host：8081或192.168.2.100：8081查看是否有该界面

flink集群部署,大数据,flink,scala,hadoop

若有，则表明flink-standalone模式部署完成

1.12 使用集群模式去验证flink-standalone是否能用

…/bin/start-scala-shell.sh remote 192.168.2.100 8081
flink集群部署,大数据,flink,scala,hadoop

1.13运行如下代码：

Scala> val text = benv.fromElements(
  "To be, or not to be,--that is the question:--",
  "Whether 'tis nobler in the mind to suffer",
  "The slings and arrows of outrageous fortune",
  "Or to take arms against a sea of troubles,")
Scala> val counts = text.flatMap { _.toLowerCase.split("\\W+") .map { (_, 1) }.groupBy(0).sum(1)
Scala> counts.print()

与下图相同则表示配置成功
flink集群部署,大数据,flink,scala,hadoop