赞
踩
咱们前面在idea中运行Flink程序的方式就是开发模式.
Flink中的Local-cluster(本地集群)模式,主要用于测试, 学习.
local-cluster模式基本属于零配置.
1.传Flink的安装包flink-1.12.0-bin-scala_2.11.tgz到hadoop162
2.解压
tar -zxvf flink-1.12.0-bin-scala_2.11.tgz -C /opt/module
3.进入目录/opt/module, 复制flink-local
cd /opt/module
cp -r flink-1.12.0 flink-local
1.打包idea中的应用
2.把不带依赖的jar包上传到目录/opt/module/flink-local下
3.启动本地集群
bin/start-cluster.sh
4.在hadoop162中启动netcat
nc -lk 9999
注意: 如果没有安装netcat需要先安装:
sudo yum install -y nc
5.命令行提交Flink应用
bin/flink run -m hadoop162:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
6.在浏览器中查看应用执行情况 http://hadoop162:8081
7.也可以在log日志查看执行结果
cat flink-atguigu-taskexecutor-0-hadoop162.out
8.也可以在WEB UI提交应用
Standalone模式又叫独立集群模式.
1.复制flink-standalone
cp -r flink-1.12.0 flink-standalone
2.修改配置文件:flink-conf.yaml
jobmanager.rpc.address: hadoop162
3.修改配置文件:workers
hadoop163
hadoop164
4.分发flink-standalone到其他节点
1.启动standalone集群
bin/start-cluster.sh
2.命令行提交Flink应用
bin/flink run -m hadoop162:8081 -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
3.查看执行情况与本地集群一致.
4. 也支持Web UI界面提交Flink应用
任何时候都有一个 主 JobManager 和多个备用 JobManagers,以便在主节点失败时有备用 JobManagers 来接管集群。这保证了没有单点故障,一旦备 JobManager 接管集群,作业就可以正常运行。主备 JobManager 实例之间没有明显的区别。每个 JobManager 都可以充当主备节点。
1.修改配置文件: link-conf.yaml
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop162:8020/flink/standalone/ha
high-availability.zookeeper.quorum: hadoop162:2181,hadoop163:2181,hadoop164:2181
high-availability.zookeeper.path.root: /flink-standalone
high-availability.cluster-id: /cluster_atguigu
2.修改配置文件: masters
hadoop162:8081
hadoop163:8081
3.分发修改的后配置文件到其他节点
4. 在/etc/profile.d/my.sh中配置环境变量
export HADOOP_CLASSPATH=`hadoop classpath`
注意:
需要提前保证HAOOP_HOME环境变量配置成功
分发到其他节点
5. 首先启动dfs集群和zookeeper集群
6. 启动standalone HA集群
bin/start-cluster.sh
7.可以分别访问
http://hadoop162:8081
http://hadoop163:8081
8. 在zkCli.sh中查看谁是leader
get /flink-standalone/cluster_atguigu/leader/rest_server_lock
杀死hadoop162上的Jobmanager, 再看leader
注意: 不管是不是leader从WEB UI上看不到区别, 并且都可以与之提交应用.
独立部署(Standalone)模式由Flink自身提供计算资源,无需其他框架提供资源,这种方式降低了和其他第三方资源框架的耦合性,独立性非常强。但是你也要记住,Flink主要是计算框架,而不是资源调度框架,所以本身提供的资源调度并不是它的强项,所以还是和其他专业的资源调度框架集成更靠谱,所以接下来我们来学习在强大的Yarn环境中Flink是如何使用的。(其实是因为在国内工作中,Yarn使用的非常多)
把Flink应用提交给Yarn的ResourceManager, Yarn的ResourceManager会申请容器从Yarn的NodeManager上面. Flink会创建JobManager和TaskManager在这些容器上.Flink会根据运行在JobManger上的job的需要的slot的数量动态的分配TaskManager资源
1.复制flink-yarn
cp -r flink-1.11.2 flink-yarn
2.配置环境变量HADOOP_CLASSPATH, 如果前面已经配置可以忽略.
在/etc/profile.d/my.sh中配置
export HADOOP_CLASSPATH=`hadoop classpath`
1.启动hadoop集群(hdfs, yarn)
2.运行无界流
bin/flink run -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
3.在yarn的ResourceManager界面查看执行情况
Flink提供了yarn上运行的3模式,分别为Application Mode, Session-Cluster和Per-Job-Cluster模式。
Session-Cluster
Session-Cluster模式需要先启动Flink集群,向Yarn申请资源, 资源申请到以后,永远保持不变。以后提交任务都向这里提交。这个Flink集群会常驻在yarn集群中,除非手工停止。
在向Flink集群提交Job的时候, 如果资源被用完了,则新的Job不能正常提交.
缺点: 如果提交的作业中有长时间执行的大作业, 占用了该Flink集群的所有资源, 则后续无法提交新的job.
所以, Session-Cluster适合那些需要频繁提交的多个小Job, 并且执行时间都不长的Job.
Per-Job-Cluster
一个Job会对应一个Flink集群,每提交一个作业会根据自身的情况,都会单独向yarn申请资源,直到作业执行完成,一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager,按需接受资源申请;适合规模大长时间运行的作业。
每次提交都会创建一个新的flink集群,任务之间互相独立,互不影响,方便管理。任务执行完成之后创建的集群也会消失。
Application Mode
Application Mode会在Yarn上启动集群, 应用jar包的main函数(用户类的main函数)将会在JobManager上执行. 只要应用程序执行结束, Flink集群会马上被关闭. 也可以手动停止集群.
与Per-Job-Cluster的区别: 就是Application Mode下, 用户的main函数是在集群中执行的
官方建议:
出于生产的需求, 我们建议使用Per-job or Application Mode,因为他们给应用提供了更好的隔离!
bin/flink run -d -t yarn-per-job -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
1.启动一个Flink-Session
bin/yarn-session.sh -d
2.在Session上运行Job
bin/flink run -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
会自动找到你的yarn-session启动的Flink集群.也可以手动指定你的yarn-session集群:
bin/flink run -t yarn-session -Dyarn.application.id=application_XXXX_YY ./flink-prepare-1.0-SNAPSHOT.jar
注意: application_XXXX_YY 指的是在yarn上启动的yarn应用
bin/flink run-application -t yarn-application -c com.atguigu.flink.java.chapter_2.Flink03_WC_UnBoundedStream ./flink-prepare-1.0-SNAPSHOT.jar
Yarn模式的高可用和Standalone模式的高可用原理不一样.
Standalone模式中, 同时启动多个Jobmanager, 一个为leader其他为standby, 当leader挂了, 其他的才会有一个成为leader.
yarn的高可用是同时只启动一个Jobmanager, 当这个Jobmanager挂了之后, yarn会再次启动一个, 其实是利用的yarn的重试次数来实现的高可用.
1.在yarn-site.xml中配置
<property>
<name>yarn.resourcemanager.am.max-attempts</name>
<value>4</value>
<description>
The maximum number of application master execution attempts.
</description>
</property>
注意: 配置完不要忘记分发, 和重启yarn
2. 在flink-conf.yaml中配置
yarn.application-attempts: 3
high-availability: zookeeper
high-availability.storageDir: hdfs://hadoop162:8020/flink/yarn/ha
high-availability.zookeeper.quorum: hadoop162:2181,hadoop163:2181,hadoop164:2181
high-availability.zookeeper.path.root: /flink-yarn
3.启动yarn-session
4.杀死Jobmanager, 查看他的复活情况
注意: yarn-site.xml中是它活的次数的上限, flink-conf.xml中的次数应该小于这个值.
scala 交互环境.
1.local模式启动 REPL
/opt/module/flink-local » bin/start-scala-shell.sh local
2.yarn-session 模式启动
先启动一个yarn-session, 然后就可以把shell跑在yarn-session上了
bin/start-scala-shell.sh yarn
Mesos是Apache下的开源分布式资源管理框架,它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署,但是在国内,依然使用着传统的Hadoop大数据框架,所以国内使用mesos框架的并不多,这里我们就不做过多讲解了。
容器化部署时目前业界很流行的一项技术,基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是Kubernetes(k8s),而Flink也在最近的版本中支持了k8s部署模式。这里我们也不做过多的讲解.
赞
踩
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。