Flink实时大数据处理技术 课件 03章.环境搭建和应用部署

第三章环境搭建和应用部署Flink实时大数据处理技术教学内容第一节Flink开发环境搭建第二节Flink词频统计第三节Flink集群部署第四节YARN模式的部署与应用提交教学目标知识目标了解Flink开发环境搭建流程了解Flink集群部署和应用提交技能目标能掌握Flink的项目搭建方法，完成词频统计能掌握Flink集群部署，完成YARN模式的部署与应用提交毕业要求 1.工程知识环境搭建和应用部署–知识点预览#节知识点难点重点应用1C03-01Flink开发环境搭建1、CentOS环境准备√

2、Flink项目搭建√3、Flink项目结构和依赖√在CentOS中安装Flink需要先安装JDK。建议安装1.8及以上版本的JDK，并将其配置到环境变量中。JDK下载地址：/java/technologies/downloads/#java8CentOS环境准备——安装JDKCentOS环境准备——快速安装FlinkFlink官网下载所需版本的二进制文件Flink下载地址：/downloads.html#flink选择flink-1.15.0-bin-scala_2.12.tgz文件下载CentOS环境准备——安装netcatNetcat是一款功能强大的网络工具，它可以通过命令行实现TCP/IP协议族的连接、传输和监听等操作。Flink中，可以使用netcat命令来模拟数据源，以方便进行测试和调试。在CentOS系统中安装netcat软件的步骤如下：1）执行以下命令以更新系统软件包：yumupdate2）使用yum命令安装netcat：yuminstallnc3）输入y确认安装。项目初始化环境准备完毕以后，本地通过Idea创建Maven项目，用于开发flink项目。打开Idea，选择"CreateNewProject"，在弹出的窗口中选择"Maven"，并勾选Createfromarchetype选项项目初始化创建Flink项目时，可以使用MavenArchetype为模板。Flink官方提供了许多不同类型的Archetype，包括Scala和Java版本的Archetype。在MVNRepository仓库中，访问地址：/，在页面中搜索查询Flinkquickstart项目模板选择1.15.0版本项目初始化可以查看到1.15.0版本对应的Maven坐标信息<!--https://mvnrepository.com/artifact/org.apache.flink/flink-quickstart-scala--><dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-quickstart-scala</artifactId> <version>1.15.0</version></dependency>在IDEA中集成上述Maven项目模板，点击AddArchetype按钮，在弹出的输入框中输入以下信息项目初始化选中flink-quickstart-scala，然后点击NEXT按钮项目名称后，安装提示完成Flink项目的创建项目结构与依赖创建完成后会自动生成相应的项目结构其中DataStreamJob是流处理的示例代码packageorg.exampleimportorg.apache.flink.streaming.api.scala._objectDataStreamJob{defmain(args:Array[String]){//设置执行环境，这是构建Flink应用程序的主入口点。valenv=StreamExecutionEnvironment.getExecutionEnvironment/**在这里，你可以开始为Flink创建执行计划。**从环境中获取一些数据开始，比如* env.fromSequence(1,10);*然后，使用操作如* .filter()* .flatMap()* .window()* .process()**等等来转换得到的DataStream<Long>。*参考编程指南：**/flink/flink-docs-stable/*///执行程序，开始计算。env.execute("FlinkScalaAPISkeleton")}}项目结构与依赖在所生成项目的POM文件中提供了以下核心配置信息，其中包括Maven属性和Flink相关的依赖项。<properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><flink.version>1.15.0</flink.version><target.java.version>1.8</target.java.version><scala.binary.version>2.12</scala.binary.version><scala.version>2.12.7</scala.version><log4j.version>2.17.1</log4j.version></properties><dependencies><!--ApacheFlink依赖--><!--这个依赖被标记为"provided"，意味着它在编译和测试时是必需的，但不会被包含在最终的JAR文件中--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version><scope>provided</scope></dependency></dependencies>Flink词频统计–知识点预览#节知识点难点重点应用1C03-02Flink词频统计1、批处理方式√

2、流处理方式√3、两种方式对比√4、向集群提交作业√批处理方式编写一个简单的词频统计的案例来尝试运行Flink项目，以下以Scala语言为例。使用Idea，新建Scalaobject文件，命名为BatchJob，并加入以下代码：importorg.apache.flink.api.scala._objectBatchJob{defmain(args:Array[String]):Unit={//获取本地执行环境valbenv=ExecutionEnvironment.getExecutionEnvironment//读取文本文件，将每一行作为数据集中的一个元素valdataSet=benv.readTextFile("D:\\wordcount.txt")//将每一行转为小写并按逗号分割dataSet.flatMap{_.toLowerCase.split(",")}//过滤掉空字符串.filter(_.nonEmpty)//将每个单词映射为(单词,1)的形式.map{(_,1)}//按照单词分组，并对每组的第二个元素（即每个单词出现的次数）求和.groupBy(0).sum(1)//将结果打印出来.print()}}批处理方式程序会读取D盘中的wordcount.txt文件，并将文件中的单词按照逗号分隔进行计数，wordcount.txt中的内容如下：a,a,a,a,ab,b,bc,cd,d程序会读取D盘中的wordcount.txt文件，并将文件中的单词按照逗号分隔进行计数，wordcount.txt中的内容如下：(d,2)(a,5)(b,3)(c,2)批处理方式这段程序是一个简单的批处理任务，它读取指定路径下的一个文本文件，并统计其中每个单词出现的次数。具体来说，程序执行了以下几个步骤：1）获取本地执行环境；2）读取指定路径下的文本文件，将每一行作为数据集中的一个元素；3）将每一行转为小写并用逗号分割，生成单词序列；4）过滤掉单词序列中的空字符串；5）将每个单词映射为(单词,1)的形式；6）按照单词分组，并对每组的第二个元素（即每个单词出现的次数）求和；7）将结果打印出来。上述过程中，FlinkAPI的Scala包提供了一些方便的操作，例如flatMap、filter、map、groupBy和sum等。在程序的最后，使用print方法将结果输出到控制台上。流处理方式在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要保持一个监听事件的状态，持续地处理捕获的数据。为了模拟这种场景，我们就不再通过读取文件来获取数据了，而是监听数据发送端主机的指定端口，统计发送来的文本数据中出现过的单词的个数。在项目中已有的DataStreamJob文件中加入以下代码:importorg.apache.flink.streaming.api.scala._objectDataStreamJob{defmain(args:Array[String]){valenv=StreamExecutionEnvironment.getExecutionEnvironmentvalstream=env.socketTextStream("00",9999)valwordCounts=stream.flatMap(line=>line.split("\\s+")).map(word=>(word,1)).keyBy(0).sum(1)wordCounts.print()env.execute("WordCount")}}流处理方式运行Main方法后，程序会处于等待状态，监听指定端口的数据，接着，切换到Linux系统，使用以下命令在linux系统中开启端口服务，作为数据源，持续发送数据。运行命令后，控制台会处于等待状态，此时，可以输入测试数据，即可观察到流处理程序的处理情况，假设我们在其中不断发送文本数据，数据格式如下:运行上面的程序，并监听发送端主机的指定端口，程序的输出如下所示。nc-lk9999helloworldhelloflinkflinkisgreat9>(world,1)5>(hello,1)13>(flink,1)5>(hello,2)13>(flink,2)16>(is,1)11>(great,1)流处理方式运行上面的程序后，输出结果前面的数字是Flink中每个并行任务（subtask）的编号，也称为subtaskindex。在Flink中，一个算子（operator）的处理逻辑可能会被分配到多个并行任务中运行，每个任务独立地处理其分配到的输入数据。在输出结果时，数据流的不同子流可能以不同的顺序到达下游算子或输出，因此输出结果的顺序可能会有所不同，但每个子任务的结果顺序始终是固定的。如果是在本地运行Flink程序，它的并行度和系统的CPU核数有关系。在本地模式的默认情况下，Flink会尝试使用本机的所有CPU核心，然后将并行任务分配到这些核心上。这个默认的并行度等于可用CPU核心数。流处理方式上面的流处理方式程序执行了以下几个步骤：1）创建流式执行环境；2）从指定的主机和端口号上获取数据流3）对流进行单词切割和计数flatMap()操作将每行文本按空格切割成单词并扁平化；map()操作将每个单词转换成一个二元组（单词，1）的形式；keyBy()操作将单词作为Key；sum()操作对每个Key对应的值求和，即对每个单词计数。4）将结果打印出来；5）启动流式执行环境。两种方式对比Flink有两种计算模式：批处理和流处理。这两种模式的最大区别在于它们处理数据的方式不同。批处理处理的是静态的有限的数据集。通常，数据集的大小是在作业启动时已知的，并且数据是一次性读取的。在批处理中，作业会把整个数据集读入内存中，然后对其进行处理，最后将结果写回外部存储系统。Flink在批处理中使用ExecutionEnvironment作为程序的入口点，批处理作业使用的是DataSetAPI。流处理处理的是动态的、无限的数据流。数据流是不断到来的，而且处理时通常只处理最近一段时间内的数据，而不是全部数据。在流处理中，作业会持续不断地接收数据流，并对其进行实时计算和转换，然后将结果输出到外部存储系统。Flink在流处理中使用StreamExecutionEnvironment作为程序的入口点，流处理作业使用的是DataStreamAPI。对于Flink而言，流才是整个处理逻辑的底层核心，所以流处理与批处理统一之后的DataStreamAPI更加强大，可以直接处理批处理和流处理的所有场景，Flink本身是流处理与批处理统一的处理架构，批量的数据集本质上也是流，没有必要用两套不同的API来实现。DataSetAPI在Flink1.12版本时被标记为“软弃用”（softdeprecated），官方建议用户尽可能使用DataStreamAPI进行开发。在Flink1.14版本中，DataSetAPI被进一步标记为“弃用”（deprecated），未来的版本中可能会完全移除该API。因此，建议用户尽快将代码从DataSetAPI迁移到DataStreamAPI。在实际应用中我们只要维护一套DataStreamAPI就可以了。两种方式对比向集群提交作业——打包程序上面我们编写了基于Flink的WordCount程序，现在我们将已经写好并且在本地测试中没有问题的程序进行编译打包后提交到Flink集群上面运行。可以在Idea右侧Maven选项中找到package按钮，双击即可对当前项目进行打包。打包生成的文件一般位于target文件夹下，其中包括了程序依赖的jar包和可执行的jar包。向集群提交作业——部署程序Flink提交应用程序有多种方式，具体如下：使用Flink自带的命令行工具；使用FlinkWebUI；使用RESTAPI；使用客户端API；需要注意的是，不同的提交方式适用于不同的场景和需求。比如，使用命令行工具和WebUI可以方便地提交简单的应用程序，而使用RESTAPI和客户端API则适用于需要动态生成应用程序拓扑结构或根据外部事件触发应用程序执行的情况。接下来主要演示前两种方式提交程序。向集群提交作业——WebUI方式部署程序打包完成后，可以将打包好的jar包上传至Flink的WEBUI页面进行提交。在左侧导航栏点击“SubmitNewJob”选项，然后点击“AddNew”按钮，选择要上传运行的jar包。在弹出的窗口中，选择“Upload”选项卡，点击“ChooseFile”按钮选择打包好的jar包文件，然后点击“Upload”按钮进行上传向集群提交作业——WebUI方式部署WebUI界面中提供了多个参数供用户设置，下面是一些常用参数的解释：EntryClass：应用程序入口类，即包含main()函数的类。Programarguments：应用程序参数，用于传递给main()函数。Parallelism：指定任务并行度，即Flink集群中可用的并行任务数。SavepointPath：指定保存点的路径，如果设置了保存点，则需要指定保存点的路径。配置完成后，切换到Linux系统，需要先使用以下命令在linux系统中开启端口服务，作为数据源，持续发送数据。nc-lk9999向集群提交作业——WebUI方式部署点击按钮“Submit”，将任务提交到集群运行。提交后，可以在左侧边栏的“RunningJobs”选项中查看正在运行的项目。向集群提交作业——WebUI方式部署输入测试信息后，可以在左侧的“TaskManagers”选项中查看输出的结果信息。在该页面上，可以看到每个任务管理器（TaskManager）的运行状态、堆栈信息、运行任务的数量、数据缓存的使用情况等信息。向集群提交作业——通过命令行部署WEBUI界面提交任务之外，也可以直接通过命令的方式来提交程序到集群，需要先将程序jar包上传至Linux系统中，此处将之前打包的程序上传至/opt/server目录下。./flinkrun-corg.example.DataStreamJob/opt/server/test-1.0-SNAPSHOT.jarFlink的run命令有很多可选参数，其中一些常用的参数包括：-c<class>或--class<class>:指定主类，用于执行应用程序的入口点。-m<host:port>或--jobmanager<host:port>:指定JobManager的地址，如果没有指定则使用默认值localhost:8081。-p<parallelism>或--parallelism<parallelism>:指定应用程序的并行度，即任务槽数量。-d或--detached:在后台运行Flink作业。-yn<number-of-taskmanagers>或--taskmanagers<number-of-taskmanagers>:指定TaskManager的数量。-ys<slots-per-taskmanager>或--slots<slots-per-taskmanager>:指定每个TaskManager的任务槽数量。-yid<YarnApplicationID>或--yarnid<YarnApplicationID>:在YARN上运行时，指定YARN应用程序的ID。-ytm<YarnTaskManagerMemory>或--yarnTaskManagerMemory<YarnTaskManagerMemory>:在YARN上运行时，指定每个TaskManager的内存大小。-yjm<YarnJobManagerMemory>或--yarnJobManagerMemory<YarnJobManagerMemory>:在YARN上运行时，指定JobManager的内存大小。Flink集群部署–知识点预览#节知识点难点重点应用1C03-03Flink集群部署1、Flink集群核心组件√

2、部署模式√3、Standalone模式√4、YARN模式√Flink集群核心组件客户端提交Flink应用程序的代码时，客户端首先将其转换为作业图（JobGraph），然后提交给作业管理器（JobManager）。Flink运行时由两种类型的进程组成：一个JobManager和一个或多个任务管理器（TaskManager）。JobManager主要负责应用程序的分布式执行、集群任务管理和调度，是控制应用程序执行的主进程，也就是说，每个应用程序都由一个不同的JobManager控制执行。Flink集群核心组件TaskManager是Flink中的工作进程，也可以称之为Worker，用于执行作业流中的任务。JobManager将工作分配到TaskManager上，在TaskManager上执行具体操作，如数据读取、转换等。TaskManager同时负责缓存和交换数据流。由于Flink是用Java和Scala实现的，因此所有组件都会在Java虚拟机（JVMs）上运行。Flink集群核心组件Flink各组件作用列表（一）组件作用实现的应用Flink客户端客户端会将批处理或流处理程序编译为数据流图，然后提交给JobManager。Flink命令行界面REST接口SQL客户端Python交互环境JobManagerJobManager是Flink集群中任务管理和调度的核心组件，根据不同的资源提供者，JobManager在高可用性、资源分配方式和支持的作业提交模式方面有所不同。JobManager的提交模式有以下几种：1.应用模式：专门为一个应用程序运行集群，作业的主方法（或客户端）在JobManager上执行。在一个应用程序中多次调用execute/executeAsync是被支持的。2.单作业模式：专门为一个作业运行集群，作业的主方法（或客户端）仅在创建集群之前运行。3.会话模式：一个JobManager实例管理多个共享同一组TaskManager的作业。

TaskManagerTaskManagers是实际执行Flink作业的组件。

Flink集群核心组件Flink各组件作用列表（二）组件作用实现的应用高可用服务Flink的JobManager可以在高可用性模式下运行，允许Flink从JobManager故障中恢复。为了更快地进行故障切换，可以启动多个备用JobManager作为备份。ZookeeperKubernetesHA资源调度Flink可以通过不同的资源调度框架进行部署，例如Kubernetes或YARN。Standalone（Flink的基本模式，只需启动Java虚拟机即可）KubernetesYARN持久化存储Flink依赖的外部存储系统，用于检查点的存储与恢复

指标存储Flink提供的Metrics可以在Flink内部收集一些指标，通过这些指标让开发人员更好地理解作业或集群的状态，通过外部存储系统如HDFS、Kafka可以存储这些指标数据。

应用级DataSource和Sink应用级DataSource和Sink不是Flink集群的内置组件，但在生产环境下常常与Flink结合使用，提高系统的性能。ApacheKafkaAmazonS3ElasticSearchApacheCassandra部署模式Flink是一个分布式系统，需要有效地分配和管理计算资源才能运行流处理程序。为此，它集成了所有常见的集群资源管理器，例如HadoopYARN，同时也可以独立运行。不同的应用场景可能需要特定的计算资源交互方式和集群资源分配方式，因此Flink为各种场景提供了不同的运行模式，这些运行模式提供了灵活的选择，以满足特定场景下的需求，主要有以下三种：会话模式（SessionMode）单作业模式（Per-JobMode）应用模式（ApplicationMode）这三种模式的主要区别在于集群的生命周期和资源的分配方式，以及应用程序的main方法在哪里执行。Standalone模式Flink的Standalone模式是Flink最简单的部署模式，所需要的所有Flink组件，都只是操作系统上运行的一个JVM进程，独立模式是独立运行的，不依赖任何外部的资源管理平台。该模式主要特点如下：1）Flink会启动一个JobManager和一个或多个TaskManager进程。这些进程可以在同一台机器上运行,也可以在不同的机器上分布式运行。2）JobManager是Flink应用程序的主进程,负责协调和调度程序的执行，一个Flink应用程序只能有一个JobManager。3）TaskManager执行用户程序的子任务,每个TaskManager最多可以运行一个或多个子任务。4）JobManager和TaskManager之间通过Akka通信。5）Flink会在配置的每个机器上启动一个ResourceManager,ResourceManager负责分配slots（任务槽）给TaskManager,并监控TaskManager的健康状态。6）如果JobManager失效,Standalone模式没有自动故障转移机制,需要手动重启JobManager。7）如果TaskManager失效,JobManager会识别并重新部署其slots。8）Standalone模式最适合本地开发和调试,以及小规模的生产部署。YARN模式YARN（YetAnotherResourceNegotiator）是Hadoop生态系统的资源管理器。YARN将集群资源划分为多个容器（Container），每个容器包含一定数量的内存、CPU和磁盘等资源。在Flink中，可以使用YARN模式在YARN集群上运行Flink应用程序，在YARN模式下，可以将Flink作业提交到YARN集群中，YARN将负责分配容器并启动FlinkJobManager和TaskManager，可以在FlinkWebUI中监视作业的执行情况，并使用Flink命令行工具提交、管理和停止作业。在Flink的YARN模式中，可以以三种方式启动Flink集群：单作业模式、会话模式、应用模式。YARN模式在单作业模式下，Flink集群在YARN集群上启动，然后运行一次性的Flink应用程序，该应用程序完成后，Flink集群将关闭。在会话模式下，Flink集群在YARN集群上启动，并一直运行直到用户显式关闭它。在会话模式下，用户可以通过Flink命令行工具提交和管理多个Flink应用程序。在应用模式下，与单作业模式类似，只不过应用程序的main()方法是在集群上执行，而不是客户端。Flink的job如何提交到YARN集群的流程如下所示：YARN模式部署与应用提交–知识点预览#节知识点难点重点应用1C03-04YARN模式的部署与应用提交1、环境准备√

2、会话模式√3、单作业模式√4、应用模式√√环境准备在将Flink任务部署至YARN集群之前，需要安装Hadoop，保证Hadoop版本至少在2.2以上，并启动HDFS及YARN组件。此次将采用伪分布模式搭建Hadoop，版本使用hadoop3.3.0，在一台主机模拟多主机，用于后续的程序开发，接下来详细介绍Hadoop在CentOS系统的伪分布式搭建方法。环境准备-配置免密登录Hadoop需要在不同的节点之间进行SSH连接。在这种情况下，免密登录是必要的。免密登录可以确保Hadoop不需要手动输入密码就能够在不同的节点之间进行SSH连接，这样可以大大简化管理员的工作。免密登录还可以提高集群的安全性，因为管理员可以通过SSH密钥来控制不同节点之间的访问权限。配置免密登录方式如下：1.配置映射，配置ip地址和主机名映射，后面使用ip的地方均可以使用主机名代替，方便管理2.生成公钥私钥3.授权，进入~/.ssh目录下，查看生成的公钥和私钥，并将公钥写入到授权文件vim/etc/hosts#文件末尾增加00rverssh-keygen-trsacd~/.sshcatid_rsa.pub>>authorized_keyschmod600authorized_keys环境准备-配置Hadoop1.下载解压Hadoop，访问/dist/hadoop/core/hadoop-3.3.0/下载Hadoop安装包。2.解压Hadoop至/opt/server目录tar-zvxfhadoop-3.3.0.tar.gz-C/opt/server/环境准备-配置Hadoop3.Hadoop的配置文件进行必要的修改，配置文件位于/opt/server/hadoop-3.3.0/etc/hadoop目录，Hadoop的主要配置文件如下：1.core-site.xml：该配置文件包含了Hadoop核心组件的通用配置选项，例如Hadoop的默认文件系统、数据复制、数据块大小。2.hdfs-site.xml：该配置文件包含了Hadoop分布式文件系统(HDFS)的配置选项，例如数据块复制数量、NameNode的数据存储路径、SecondaryNameNode的备份间隔时间等配置项。3.yarn-site.xml：该配置文件包含了Hadoop资源管理器(YARN)的配置选项，例如NodeManager的数量、ResourceManager的资源分配方式、作业调度器的类型等配置项。4.mapred-site.xml：该配置文件包含了MapReduce框架的配置选项，例如作业跟踪器的数量、任务分配器的类型、作业提交和执行的队列等配置项。5.hadoop-env.sh：该配置文件包含了Hadoop环境变量的设置，例如JAVA_HOME、HADOOP_HOME等变量的设置。4.具体的步骤如下：step1：

2.解压Hadoop至/opt/server目录vimhadoop-env.sh#使用G+o，跳转到文件最后一行，并插入exportJAVA_HOME=/opt/server/jdk1.8.0_131exportHDFS_NAMENODE_USER=rootexportHDFS_DATANODE_USER=rootexportHDFS_SECONDARYNAMENODE_USER=rootexportYARN_RESOURCEMANAGER_USER=rootexportYARN_NODEMANAGER_USER=root环境准备-配置Hadoop<configuration><property><!--设置默认使用的文件系统Hadoop支持file、HDFS、GFS、ali|Amazon云等文件系统--><name>fs.defaultFS</name><value>hdfs://server:8020</value></property><property><!--设置Hadoop本地保存数据路径，此目录不需要提前创建会自动生成--><name>hadoop.tmp.dir</name><value>/home/hadoop/data</value></property><property><!--设置HDFSwebUI用户身份--><name>hadoop.http.staticuser.user</name><value>root</value></property></configuration>step2：修改core-site.xml文件，分别指定hdfs协议文件系统的通信地址及hadoop存储临时文件的目录（此目录不需要手动创建）环境准备-配置Hadoopstep3：修改hdfs-site.xml，指定dfs的副本系数<configuration>

<property>

<!--由于我们这里搭建是单机版本，所以指定dfs的副本系数为1-->

<name>dfs.replication</name>

<value>1</value>

</property></configuration>step4：修改workers文件，配置所有从属节点#配置所有从属节点的主机名或IP地址，由于是单机版本，所以指定本机即可：serverstep5：修改workers文件，配置所有从属节点cd/opt/server/hadoop-3.3.0/bin./hdfsnamenode-format环境准备-配置Hadoopstep6：启动HDFS，进入/opt/server/hadoop-3.3.0/sbin/目录下，启动HDFS./start-dfs.shstep7：验证是否启动，执行jps查看NameNode和DataNode服务是否已经启动[root@serverbin]#jps41032DataNode41368Jps40862NameNode41246SecondaryNameNode环境准备-配置Hadoopstep8：进入/opt/server/hadoop-3.3.0/etc/hadoop目录下，修改mapred-site.xml文件<configuration><!--设置yarn为MapReduce的执行框架--><property><name></name><value>yarn</value></property><!--MR程序历史服务器web端地址--><property><name>mapreduce.jobhistory.webapp.address</name><value>server:19888</value></property><property><name>yarn.app.mapreduce.am.env</name><value>HADOOP_MAPRED_HOME=${HADOOP_HOME}</value></property><property><name>mapreduce.map.env</name><value>HADOOP_MAPRED_HOME=${HADOOP_HOME}</value></property><property><name>mapreduce.reduce.env</name><value>HADOOP_MAPRED_HOME=${HADOOP_HOME}</value></property></configuration>环境准备-配置Hadoopstep9：修改yarn-site.xml文件，配置NodeManager上运行的附属服务<configuration><property><!--配置NodeManager上运行的附属服务。需要配置成mapreduce_shuffle后才可以在Yarn上运行MapReduce程序。--><name>yarn.nodemanager.aux-services</name><value>mapreduce_shuffle</value></property><!--是否对容器实施物理内存限制--><property><name>yarn.nodemanager.pmem-check-enabled</name><value>false</value></property><!--是否对容器实施虚拟内存限制。--><property><name>yarn.nodemanager.vmem-check-enabled</name><value>false</value></property><!--历史日志保存的时间，单位是秒，当前设置的值为7天--><property><name>yarn.log-aggregation.retain-seconds</name><value>604800</value></property></configuration>环境准备-配置Hadoopstep10：进入/opt/server/hadoop-3.3.0/sbin/目录下，启动YARN./start-yarn.shstep11：验证是否启动成功，执行jps命令查看NodeManager和ResourceManager服务是否已经启动[root@serverbin]#jps41655ResourceManager41032DataNode42125Jps40862NameNode41246SecondaryNameNode41983NodeManagerstep12：启动历史日志服务，脚本位于sbin目录，开启此服务后，可以在Web控制台查看MR运行的日志信息./mr-jobhistory-daemon.shstarthistoryserver至此，Hadoop的伪分布式环境已经搭建成功。环境准备-Flink与Yarn集成编