Flink on Yarn三部曲之三:提交Flink任务

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Flink on Yarn在使用的时候分为两种模式,Job Mode和Session Mode,一起来体验这两种模式

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码): https://github.com/zq2599/blog_demos

本篇概览

  • 本文是《Flink on Yarn三部曲》系列的终篇,先简单回顾前面的内容:
  1. 《Flink on Yarn三部曲之一:准备工作》:准备好机器、脚本、安装包;
  2. 《Flink on Yarn三部曲之二:部署和设置》:完成CDH和Flink部署,并在管理页面做好相关的设置;
  • 现在Flink、Yarn、HDFS都就绪了,接下来实践提交Flink任务到Yarn执行;

两种Flink on YARN模式

  • 实践之前,对Flink on YARN先简单了解一下,如下图所示,Flink on Yarn在使用的时候分为两种模式,Job ModeSession Mode

在这里插入图片描述

  • Session Mode:在YARN中提前初始化一个Flink集群,以后所有Flink任务都提交到这个集群,如下图:

在这里插入图片描述

  • Job Mode:每次提交Flink任务都会创建一个专用的Flink集群,任务完成后资源释放,如下图:

在这里插入图片描述

  • 接下来分别实战这两种模式;

准备实战用的数据(CDH服务器)

  • 接下来提交的Flink任务是经典的WordCount,先在HDFS中准备一份文本文件,后面提交的Flink任务都会读取这个文件,统计里面每个单词的数字,准备文本的步骤如下:
  1. SSH登录CDH服务器;
  2. 切换到hdfs账号:su - hdfs
  3. 下载实战用的txt文件:
wget https://github.com/zq2599/blog_demos/blob/master/files/GoneWiththeWind.txt
  1. 创建hdfs文件夹:hdfs dfs -mkdir /input
  2. 将文本文件上传到/input目录:hdfs dfs -put ./GoneWiththeWind.txt /input
  • 准备工作完成,可以提交任务试试了。

Session Mode实战

  • SSH登录CDH服务器;
  • 切换到hdfs账号:su - hdfs
  • 进入目录:/opt/flink-1.7.2/
  • 执行如下命令创建Flink集群,-n参数表示TaskManager的数量,-jm表示JobManager的内存大小,-tm表示每个TaskManager的内存大小:
./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024
  • 创建成功后,控制台输出如下图,注意红框中的提示,表明可以通过38301端口访问Flink:

在这里插入图片描述

  • 浏览器访问CDH服务器的38301端口,可见Flink服务已经启动:

在这里插入图片描述

  • 浏览器访问CDH服务器的8088端口,可见YARN的Application(即Flink集群)创建成功,如下图,红框中是任务ID,稍后结束Application的时候会用到此ID:

在这里插入图片描述

  • 再开启一个终端,SSH登录CDH服务器,切换到hdfs账号,进入目录:/opt/flink-1.7.2
  • 执行以下命令,就会提交一个Flink任务(安装包自带的WordCount例子),并指明将结果输出到HDFS的wordcount-result.txt文件中:
bin/flink run ./examples/batch/WordCount.jar \
-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \
-output hdfs://192.168.50.134:8020/wordcount-result.txt
  • 执行完毕后,控制台输出如下:

在这里插入图片描述

  • flink的WordCount任务结果保存在hdfs,我们将结果取出来看看:hdfs dfs -get /wordcount-result.txt
  • vi打开wordcount-result.txt文件,如下图,可见任务执行成功,指定文本中的每个单词数量都统计出来了:

在这里插入图片描述

  • 浏览器访问Flink页面(CDH服务器的38301端口),也能看到任务的详细情况:

在这里插入图片描述

  • 销毁这个Flink集群的方法是在控制台执行命令:yarn application -kill application_1580173588985_0002

在这里插入图片描述

  • Session Mode的实战就完成了,接下来我们来尝试Job Mode;

Job Mode

  • 执行以下命令,创建一个Flink集群,该集群只用于执行参数中指定的任务(wordCount.jar),结果输出到hdfs的wordcount-result-1.txt文件:
bin/flink run -m yarn-cluster \
-yn 2 \
-yjm 1024 \
-ytm 1024 \
./examples/batch/WordCount.jar \
-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \
-output hdfs://192.168.50.134:8020/wordcount-result-1.txt
  • 控制台输出如下,表明任务执行完成:

在这里插入图片描述

  • 如果您的内存和CPU核数充裕,可以立即执行以下命令再创建一个Flink集群,该集群只用于执行参数中指定的任务(wordCount.jar),结果输出到hdfs的wordcount-result-2.txt文件:
bin/flink run -m yarn-cluster \
-yn 2 \
-yjm 1024 \
-ytm 1024 \
./examples/batch/WordCount.jar \
-input hdfs://192.168.50.134:8020/input/GoneWiththeWind.txt \
-output hdfs://192.168.50.134:8020/wordcount-result-2.txt
  • 在YARN管理页面可见任务已经结束:

在这里插入图片描述

  • 执行命令hdfs dfs -ls /查看结果文件,已经成功生成:

在这里插入图片描述

  • 执行命令hdfs dfs -get /wordcount-result-1.txt下载结果文件到本地,检查数据正常;
  • 至此,Flink on Yarn的部署、设置、提交都实践完成,《Flink on Yarn三部曲》系列也结束了,如果您也在学习Flink,希望本文能够给您一些参考,也建议您根据自身情况和需求,修改ansible脚本,搭建更适合自己的环境;

欢迎关注阿里云开发者社区博客:程序员欣宸

学习路上,你不孤单,欣宸原创一路相伴...
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
1月前
|
消息中间件 资源调度 关系型数据库
如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理
本文介绍了如何在Flink on YARN环境中配置Debezium CDC 3.0,以实现实时捕获数据库变更事件并将其传输到Flink进行处理。主要内容包括安装Debezium、配置Kafka Connect、创建Flink任务以及启动任务的具体步骤,为构建实时数据管道提供了详细指导。
75 9
|
2月前
|
Java Shell Maven
Flink-11 Flink Java 3分钟上手 打包Flink 提交任务至服务器执行 JobSubmit Maven打包Ja配置 maven-shade-plugin
Flink-11 Flink Java 3分钟上手 打包Flink 提交任务至服务器执行 JobSubmit Maven打包Ja配置 maven-shade-plugin
128 4
|
2月前
|
资源调度 分布式计算 大数据
大数据-111 Flink 安装部署 YARN部署模式 FlinkYARN模式申请资源、提交任务
大数据-111 Flink 安装部署 YARN部署模式 FlinkYARN模式申请资源、提交任务
117 0
|
4月前
|
SQL Shell API
实时计算 Flink版操作报错合集之任务提交后出现 "cannot run program "/bin/bash": error=1, 不允许操作" ,是什么原因
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
实时计算 Flink版操作报错合集之任务提交后出现 "cannot run program "/bin/bash": error=1, 不允许操作" ,是什么原因
|
4月前
|
监控 Cloud Native 流计算
实时计算 Flink版产品使用问题之如何查看和管理任务
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
消息中间件 资源调度 Kafka
实时计算 Flink版操作报错合集之提交任务后,如何解决报错:UnavailableDispatcherOperationException
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
4月前
|
资源调度 Java Scala
实时计算 Flink版产品使用问题之如何实现ZooKeeper抖动导致任务失败时,能从最近的检查点重新启动任务
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
资源调度 Oracle Java
实时计算 Flink版产品使用问题之在YARN集群上运行时,如何查看每个并行度的详细处理数据情况
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
Kubernetes Java 数据库连接
实时计算 Flink版产品使用问题之部署到 Kubernetes 集群时,任务过一会儿自动被取消,该如何排查
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4月前
|
资源调度 安全 数据处理
实时计算 Flink版产品使用问题之提交任务时如何设置TaskManager的数量
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。