目录大数据的特性HDFS 读流程的基本步骤HDFS 写流程的基本步骤MapReduce的执行过程MapReduce 中 combiner 作用hadoop 调度器及其工作方法Hive 中内部表与外部表区别(创建删除角度)Hadoop 的 2 个主要组件及其功能Hadoop MapReduce 的工作流程正常工作的 hadoop 集群中 Hadoop 分别启动哪些进程及其作用Hadoop 中的 NameNode 和 DataNode 分别承担的角色Hadoop 集群中的 NameNode 如何保证单点故障问题Hadoop 中的 Block 是什么及其作用Hadoop 在大数据处理中的优势FsImage 镜像文件和 EditLog 日志文件 HDFS 中提供的 Secondary NameNode 节点的职责Hadoop 中的分区器Partitioner的作用YARN 提交作业的流程MapperReducerMaster代码大数据的特性全样而非抽样、效率而非精确 、相关而非因果HDFSHadoop Distributed File System的读流程主要涉及客户端Client、NameNode 和 DataNode 之间的交互。HDFS 读流程的基本步骤文件切分对于上传操作 当文件上传到 HDFS 时客户端Client首先将文件切分成一个个的数据块Block然后进行上传。与 NameNode 交互 客户端向 NameNode 发送请求获取要读取文件的位置信息。这个位置信息包括了组成该文件的所有数据块Block以及它们所在的 DataNode 的地址。与 DataNode 交互 一旦客户端获得了文件的数据块及其所在的 DataNode 的位置信息它就会与这些 DataNode 进行交互以读取数据。 客户端可能会并行地从多个 DataNode 读取数据块以提高读取效率。数据读取 客户端从 DataNode 读取数据块并在本地重新组装这些数据块以还原原始文件。关闭连接 当文件读取完成后客户端关闭与 DataNode 和 NameNode 的连接。HDFSHadoop Distributed File System的写流程涉及客户端Client、NameNode 和 DataNode 之间的交互。HDFS 写流程的基本步骤客户端发起请求 客户端Client首先向 HDFS 的 NameNode 发送写入文件的请求。NameNode 是 HDFS 的 主节点负责管理文件系统的命名空间和元数据信息。NameNode 响应请求 NameNode 接收到客户端的请求后会检查自身是否正常运行判断要创建的文件是否存在 以及客户端是否具有创建文件的权限。如果以上检查都通过NameNode 会在 HDFS 文件系 统中创建一个空文件并将这一操作记录在 edits.log 文件中。如果检查中有任何一项未通过 NameNode 会向客户端抛出异常文件创建失败。文件切分 客户端将待写入的文件切分成固定大小的数据块通常为 128MB。每个数据块都会被分配一个唯一的块标识符。数据块副本选择 在写入数据块之前客户端需要选择数据块的副本位置。这通常是基于 HDFS 的副本放置策略旨在减少数据传输的开销和延迟。数据块写入 客户端将数据块分别发送给副本位置所在的 DataNode。DataNode 接收到数据块后会将数据块暂存到本地磁盘上的临时文件中。数据块复制 一旦数据块被写入到一个 DataNode 的临时文件中该 DataNode 会将其复制到其他副本位置所在的 DataNode 上。副本确认 当所有副本都完成数据写入后DataNode 会向客户端发送副本确认信息。客户端收到所有副本的确认信息后将告知 NameNode 数据块的写入完成。MapReduce的执行过程MapReduce 中 combiner 作用Hadoop 中的 Combiner 是 MapReduce 作业中的一个可选组件用于在 Map 阶段对输出数据进行部分聚合。Combiner 可以减少 Map 和 Reduce 之间传输的数据量从而降低网络带宽的消耗。同时Combiner 还可以减少 Reduce 阶段的计算量提高作业的执行效率。hadoop 调度器及其工作方法Fifo schedular:默认先进先出的原则。Capacity schedular:计算能力调度器选择占用最小优先级高的先执行以此类推。Fair schedule公平调度所有的 job 具有相同的资源。Hive 中内部表与外部表区别(创建删除角度)创建表阶段 外部表创建表的时候不会移动数到数据仓库目录中/user/hive/warehouse只会记录表数 据存放的路径内部表会把数据复制或剪切到表的目录下。删除表阶段 外部表在删除表的时候只会删除表的元数据信息不会删除表数据内部表删除时会将元数据 信息和表数据同时删除Hadoop 的 2 个主要组件及其功能Hadoop 主要由两个核心组件组成Hadoop Distributed File System (HDFS) 和 MapReduce。HDFSHadoop 的分布式文件系统用于存储和管理大数据集。它采用主/从架构包括一个 NameNode主节点和多个 DataNode从节点。NameNode 管理文件系统的元数据 而 DataNode 存储实际的数据块。MapReduceHadoop 的编程框架用于处理存储在 HDFS 中的大数据。它将复杂的数据处 理任务分解为两个主要阶段Map 阶段和 Reduce 阶段。Map 阶段将数据划分为多个键值对Reduce 阶段则对这些键值对进行聚合和输出。Hadoop MapReduce 的工作流程输入阶段MapReduce 作业从 HDFS 或其他输入源读取数据。Map 阶段Map 任务将输入数据划分为多个键值对并对其进行处理。处理结果作为中间 键值对输出到本地磁盘。Shuffle 阶段MapReduce 框架将 Map 任务输出的中间键值对按照键进行排序和分组并将 相同键的值传递给 Reduce 任务。Reduce 阶段Reduce 任务接收 Shuffle 阶段传递过来的键值对对它们进行聚合处理并将 结果输出到 HDFS 或其他输出源。输出阶段MapReduce 作业将 Reduce 任务输出的结果存储到 HDFS 或其他输出目标。正常工作的 hadoop 集群中 Hadoop 分别启动哪些进程及其作用Namenode管理集群并记录 datanode 文件信息。Secondnamenode可以做冷备对一定范围内数据做快照性备份。Datanode存储数据 。ResourceManager负责集群中所有资源的统一管理和分配它接收来自各个节点 nodemanager的资源汇报信息并把这些信息按照一定的策略分配给各个应用程序实际上是 applicationmanagerNodeManager是 yarn 中每个节点上的代理它管理 hadoop 集群中单个计算节点包括与 resoucemanager 保持通信监督 container 的生命周期管理监控每个 container 的资源 使用内存、cpu 等情况追踪节点健康状况管理日志和不同应用程序用到的附属服务。Hadoop 中的 NameNode 和 DataNode 分别承担的角色NameNodeNameNode 是 HDFS 中的主节点负责维护文件系统的元数据如文件目录结 构、文件块与 DataNode 的映射关系等。它不接受客户端的读写请求而是将请求转发给 DataNode并协调数据的读写操作。DataNodeDataNode 是 HDFS 中的从节点负责存储实际的数据块。客户端通过与 NameNode 通信获取数据块的存储位置然后直接与 DataNode 进行数据的读写操作。DataNode 还会定 期向 NameNode 发送心跳信息以报告自身的状态和数据块的完整性。Hadoop 集群中的 NameNode 如何保证单点故障问题Hadoop 集群中的 NameNode 单点故障问题可以通过设置 NameNode 的高可用性HA来解 决。HA 配置使用两个 NameNode 实例一个处于活动状态Active另一个处于备用状态 Standby。当活动 NameNode 出现故障时备用 NameNode 会自动接管其工作从而确保集群的连续 性和可用性。此外还可以使用 Zookeeper 等分布式协调服务来监控和管理 NameNode 的状态和切换过程。Hadoop 中的 Block 是什么及其作用Hadoop 中的 Block 是 HDFS 中数据存储的基本单位通常大小为 128MB可配置。HDFS 将数据划分为多个 Block并将它们分散存储在集群中的多个 DataNode 上。这种设计提高了数据的可靠性和可扩展性因为每个 Block 都有多个副本存储在不同的节点上并且可以通过增加节点来扩展存储能力。Hadoop 在大数据处理中的优势可扩展性Hadoop 能够处理 PB 级甚至更大的数据集通过增加节点可以轻松地扩展集群的处理能力。容错性Hadoop 采用分布式存储和计算的方式将数据分散存储在多个节点上并通过数据冗余和故障恢复机制保证数据的高可用性。灵活性Hadoop 支持多种编程语言和工具进行数据处理如 Java、Python 等同时提供了丰富的 API 和生态系统供开发者使用。成本效益Hadoop 运行在普通硬件上相比传统的高性能计算集群具有更低的成本。同时 Hadoop 的开源特性使得用户可以免费获取和使用它。FsImage 镜像文件和 EditLog 日志文件 HDFS 中提供的 Secondary NameNode 节点的职责FsImage 镜像文件用于存储整个文件系统命名空间的信息EditLog 日志文件用于持久化记录文件系统元数据发生的变化。Secondary NameNode 节点主要是周期性的把 NameNode 中的 EditLog 日志文件合并到 FsImage 镜像文件中从而减小 EditLog 日志文件的大小缩短集群重启时间并且也保证 了 HDFS 系统的完整性。Hadoop 中的分区器Partitioner的作用在 Hadoop 的 MapReduce 作业中分区器用于确定 Map 阶段输出的键值对应该发送到哪个 Reducer 进行处理。分区器根据键的哈希值或其他逻辑将数据划分为不同的分区并将每个分区的数据发送到对应的 Reducer。这种设计可以确保具有相同键的数据被发送到同一个 Reducer 进行处理从而便于在 Reduce 阶段进行聚合操作。YARNYet Another Resource Negotiator是 Hadoop 的一个资源管理系统它负责集群中的 资源管理和作业调度。YARN 提交作业的流程作业提交 Client客户端调用 job.waitForCompletion 方法向整个集群提交 MapReduce 作业。 Client 向 ResourceManagerRM申请一个作业 ID。 RM 给 Client 返回该 job 的资源提交路径和作业 ID。 Client 提交 jar 包、切片信息和配置文件到指定的资源提交路径。 Client 提交完资源后向 RM 申请运行 MRAppMasterMapReduce 应用程序主。作业初始化 当 RM 收到 Client 的请求后将该 job 添加到容量调度器中。 某一个空闲的 NodeManagerNM领取到该 Job。 该 NM 创建 Container并产生 MRAppMaster。 下载 Client 提交的资源到本地。资源分配 MRAppMaster 与 RM 的应用程序管理器Application Manager进行交互协商以获取运行 作业所需的资源。 RM 的调度器根据集群的资源情况和配置的调度策略如容量调度、公平调度等将资源 分配给 MRAppMaster。任务执行 MRAppMaster 将获取到的资源分配给各个 NM。 NM 上的 Executor执行器根据分配到的资源和作业的具体任务开始执行 Map Task 或 Reduce Task。作业监控和完成 RM 的应用程序管理器负责监控 MRAppMaster 的运行状态并在必要时进行重启。 当所有任务执行完毕后MRAppMaster 通知 RM 作业已完成。 Client 通过 RM 获取作业完成状态并获取结果。MapperReducerMaster词频统计代码Mapperpackage edu.mapper; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class WordCountMapper extends MapperLongWritable, Text, Text, IntWritable { Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //获取读取到的每一行数据 String line value.toString(); //通过空格进行分割 String[] words line.split( ); //将初步处理的结果进行输出 for (String word: words) { context.write(new Text(word), new IntWritable(1)); } } }Reducerpackage edu.reducer; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class WordCountReducer extends ReducerText, IntWritable, Text,IntWritable { protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { Integer count 0; for(IntWritable value: values){ //将每一个分组中的数据进行累加计数 count value.get(); } //将最终结果进行输出 context.write(key, new IntWritable(count)); } }Masterpackage edu; import edu.mapper.WordCountMapper; import edu.reducer.WordCountReducer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class WordCountMaster { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { //初始化配置 Configuration conf new Configuration(); conf.set(fs.defaultFS, hdfs://virtual-machine-name:9000); //初始化job参数指定job名称 Job job Job.getInstance(conf,WordCount); //设置运行job的类 job.setJarByClass(WordCountMaster.class); //设置Mapper类 job.setMapperClass(WordCountMapper.class); //设置Reducer类 job.setReducerClass(WordCountReducer.class); //设置Map的输出数据类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); //设置Reducer的输出数据类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); //设置输入的路径 //FileInputFormat.setInputPaths(job, new Path(Hadoop_Urlinput/wordCount)); //设置输出的路径 //FileOutputFormat.setOutputPath(job, new Path(Hadoop_Urloutput/wordCount/1)); //指定该mapreduce程序数据的输入和输出路径 Path inputPath new Path(/wordcount/input); Path outputPath new Path(/wordcount/output); FileSystem fs FileSystem.get(conf); if(fs.exists(outputPath)){ fs.delete(outputPath, true); } FileInputFormat.setInputPaths(job, inputPath); FileOutputFormat.setOutputPath(job, outputPath); //提交job boolean result job.waitForCompletion(true); //执行成功后进行后续操作 if (result) { System.out.println(Congratulations!); } } }计算第四列每个元素出现的个数a,b,c,db,b,f,ea,a,c,fmapperimport java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class LogMapper extends MapperLongWritable, Text, Text, LongWritable { Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); String[] columns line.split(,); if (columns.length 4) { context.write(new Text(columns[3]), new LongWritable(1)); } } }reducerimport java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class LogReducer extends ReducerText, LongWritable, Text, LongWritable { Override protected void reduce(Text key, IterableLongWritable values, Context context) throws IOException, InterruptedException { long count 0; for (LongWritable value : values) { count value.get(); } context.write(key, new LongWritable(count)); } }mainimport org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class LogAnalysis { public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, Log Analysis); job.setJarByClass(LogAnalysis.class); job.setMapperClass(LogMapper.class); job.setReducerClass(LogReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
目录大数据的特性HDFS 读流程的基本步骤HDFS 写流程的基本步骤MapReduce的执行过程MapReduce 中 combiner 作用hadoop 调度器及其工作方法Hive 中内部表与外部表区别(创建删除角度)Hadoop 的 2 个主要组件及其功能Hadoop MapReduce 的工作流程正常工作的 hadoop 集群中 Hadoop 分别启动哪些进程及其作用Hadoop 中的 NameNode 和 DataNode 分别承担的角色Hadoop 集群中的 NameNode 如何保证单点故障问题Hadoop 中的 Block 是什么及其作用Hadoop 在大数据处理中的优势FsImage 镜像文件和 EditLog 日志文件 HDFS 中提供的 Secondary NameNode 节点的职责Hadoop 中的分区器Partitioner的作用YARN 提交作业的流程MapperReducerMaster代码大数据的特性全样而非抽样、效率而非精确 、相关而非因果HDFSHadoop Distributed File System的读流程主要涉及客户端Client、NameNode 和 DataNode 之间的交互。HDFS 读流程的基本步骤文件切分对于上传操作 当文件上传到 HDFS 时客户端Client首先将文件切分成一个个的数据块Block然后进行上传。与 NameNode 交互 客户端向 NameNode 发送请求获取要读取文件的位置信息。这个位置信息包括了组成该文件的所有数据块Block以及它们所在的 DataNode 的地址。与 DataNode 交互 一旦客户端获得了文件的数据块及其所在的 DataNode 的位置信息它就会与这些 DataNode 进行交互以读取数据。 客户端可能会并行地从多个 DataNode 读取数据块以提高读取效率。数据读取 客户端从 DataNode 读取数据块并在本地重新组装这些数据块以还原原始文件。关闭连接 当文件读取完成后客户端关闭与 DataNode 和 NameNode 的连接。HDFSHadoop Distributed File System的写流程涉及客户端Client、NameNode 和 DataNode 之间的交互。HDFS 写流程的基本步骤客户端发起请求 客户端Client首先向 HDFS 的 NameNode 发送写入文件的请求。NameNode 是 HDFS 的 主节点负责管理文件系统的命名空间和元数据信息。NameNode 响应请求 NameNode 接收到客户端的请求后会检查自身是否正常运行判断要创建的文件是否存在 以及客户端是否具有创建文件的权限。如果以上检查都通过NameNode 会在 HDFS 文件系 统中创建一个空文件并将这一操作记录在 edits.log 文件中。如果检查中有任何一项未通过 NameNode 会向客户端抛出异常文件创建失败。文件切分 客户端将待写入的文件切分成固定大小的数据块通常为 128MB。每个数据块都会被分配一个唯一的块标识符。数据块副本选择 在写入数据块之前客户端需要选择数据块的副本位置。这通常是基于 HDFS 的副本放置策略旨在减少数据传输的开销和延迟。数据块写入 客户端将数据块分别发送给副本位置所在的 DataNode。DataNode 接收到数据块后会将数据块暂存到本地磁盘上的临时文件中。数据块复制 一旦数据块被写入到一个 DataNode 的临时文件中该 DataNode 会将其复制到其他副本位置所在的 DataNode 上。副本确认 当所有副本都完成数据写入后DataNode 会向客户端发送副本确认信息。客户端收到所有副本的确认信息后将告知 NameNode 数据块的写入完成。MapReduce的执行过程MapReduce 中 combiner 作用Hadoop 中的 Combiner 是 MapReduce 作业中的一个可选组件用于在 Map 阶段对输出数据进行部分聚合。Combiner 可以减少 Map 和 Reduce 之间传输的数据量从而降低网络带宽的消耗。同时Combiner 还可以减少 Reduce 阶段的计算量提高作业的执行效率。hadoop 调度器及其工作方法Fifo schedular:默认先进先出的原则。Capacity schedular:计算能力调度器选择占用最小优先级高的先执行以此类推。Fair schedule公平调度所有的 job 具有相同的资源。Hive 中内部表与外部表区别(创建删除角度)创建表阶段 外部表创建表的时候不会移动数到数据仓库目录中/user/hive/warehouse只会记录表数 据存放的路径内部表会把数据复制或剪切到表的目录下。删除表阶段 外部表在删除表的时候只会删除表的元数据信息不会删除表数据内部表删除时会将元数据 信息和表数据同时删除Hadoop 的 2 个主要组件及其功能Hadoop 主要由两个核心组件组成Hadoop Distributed File System (HDFS) 和 MapReduce。HDFSHadoop 的分布式文件系统用于存储和管理大数据集。它采用主/从架构包括一个 NameNode主节点和多个 DataNode从节点。NameNode 管理文件系统的元数据 而 DataNode 存储实际的数据块。MapReduceHadoop 的编程框架用于处理存储在 HDFS 中的大数据。它将复杂的数据处 理任务分解为两个主要阶段Map 阶段和 Reduce 阶段。Map 阶段将数据划分为多个键值对Reduce 阶段则对这些键值对进行聚合和输出。Hadoop MapReduce 的工作流程输入阶段MapReduce 作业从 HDFS 或其他输入源读取数据。Map 阶段Map 任务将输入数据划分为多个键值对并对其进行处理。处理结果作为中间 键值对输出到本地磁盘。Shuffle 阶段MapReduce 框架将 Map 任务输出的中间键值对按照键进行排序和分组并将 相同键的值传递给 Reduce 任务。Reduce 阶段Reduce 任务接收 Shuffle 阶段传递过来的键值对对它们进行聚合处理并将 结果输出到 HDFS 或其他输出源。输出阶段MapReduce 作业将 Reduce 任务输出的结果存储到 HDFS 或其他输出目标。正常工作的 hadoop 集群中 Hadoop 分别启动哪些进程及其作用Namenode管理集群并记录 datanode 文件信息。Secondnamenode可以做冷备对一定范围内数据做快照性备份。Datanode存储数据 。ResourceManager负责集群中所有资源的统一管理和分配它接收来自各个节点 nodemanager的资源汇报信息并把这些信息按照一定的策略分配给各个应用程序实际上是 applicationmanagerNodeManager是 yarn 中每个节点上的代理它管理 hadoop 集群中单个计算节点包括与 resoucemanager 保持通信监督 container 的生命周期管理监控每个 container 的资源 使用内存、cpu 等情况追踪节点健康状况管理日志和不同应用程序用到的附属服务。Hadoop 中的 NameNode 和 DataNode 分别承担的角色NameNodeNameNode 是 HDFS 中的主节点负责维护文件系统的元数据如文件目录结 构、文件块与 DataNode 的映射关系等。它不接受客户端的读写请求而是将请求转发给 DataNode并协调数据的读写操作。DataNodeDataNode 是 HDFS 中的从节点负责存储实际的数据块。客户端通过与 NameNode 通信获取数据块的存储位置然后直接与 DataNode 进行数据的读写操作。DataNode 还会定 期向 NameNode 发送心跳信息以报告自身的状态和数据块的完整性。Hadoop 集群中的 NameNode 如何保证单点故障问题Hadoop 集群中的 NameNode 单点故障问题可以通过设置 NameNode 的高可用性HA来解 决。HA 配置使用两个 NameNode 实例一个处于活动状态Active另一个处于备用状态 Standby。当活动 NameNode 出现故障时备用 NameNode 会自动接管其工作从而确保集群的连续 性和可用性。此外还可以使用 Zookeeper 等分布式协调服务来监控和管理 NameNode 的状态和切换过程。Hadoop 中的 Block 是什么及其作用Hadoop 中的 Block 是 HDFS 中数据存储的基本单位通常大小为 128MB可配置。HDFS 将数据划分为多个 Block并将它们分散存储在集群中的多个 DataNode 上。这种设计提高了数据的可靠性和可扩展性因为每个 Block 都有多个副本存储在不同的节点上并且可以通过增加节点来扩展存储能力。Hadoop 在大数据处理中的优势可扩展性Hadoop 能够处理 PB 级甚至更大的数据集通过增加节点可以轻松地扩展集群的处理能力。容错性Hadoop 采用分布式存储和计算的方式将数据分散存储在多个节点上并通过数据冗余和故障恢复机制保证数据的高可用性。灵活性Hadoop 支持多种编程语言和工具进行数据处理如 Java、Python 等同时提供了丰富的 API 和生态系统供开发者使用。成本效益Hadoop 运行在普通硬件上相比传统的高性能计算集群具有更低的成本。同时 Hadoop 的开源特性使得用户可以免费获取和使用它。FsImage 镜像文件和 EditLog 日志文件 HDFS 中提供的 Secondary NameNode 节点的职责FsImage 镜像文件用于存储整个文件系统命名空间的信息EditLog 日志文件用于持久化记录文件系统元数据发生的变化。Secondary NameNode 节点主要是周期性的把 NameNode 中的 EditLog 日志文件合并到 FsImage 镜像文件中从而减小 EditLog 日志文件的大小缩短集群重启时间并且也保证 了 HDFS 系统的完整性。Hadoop 中的分区器Partitioner的作用在 Hadoop 的 MapReduce 作业中分区器用于确定 Map 阶段输出的键值对应该发送到哪个 Reducer 进行处理。分区器根据键的哈希值或其他逻辑将数据划分为不同的分区并将每个分区的数据发送到对应的 Reducer。这种设计可以确保具有相同键的数据被发送到同一个 Reducer 进行处理从而便于在 Reduce 阶段进行聚合操作。YARNYet Another Resource Negotiator是 Hadoop 的一个资源管理系统它负责集群中的 资源管理和作业调度。YARN 提交作业的流程作业提交 Client客户端调用 job.waitForCompletion 方法向整个集群提交 MapReduce 作业。 Client 向 ResourceManagerRM申请一个作业 ID。 RM 给 Client 返回该 job 的资源提交路径和作业 ID。 Client 提交 jar 包、切片信息和配置文件到指定的资源提交路径。 Client 提交完资源后向 RM 申请运行 MRAppMasterMapReduce 应用程序主。作业初始化 当 RM 收到 Client 的请求后将该 job 添加到容量调度器中。 某一个空闲的 NodeManagerNM领取到该 Job。 该 NM 创建 Container并产生 MRAppMaster。 下载 Client 提交的资源到本地。资源分配 MRAppMaster 与 RM 的应用程序管理器Application Manager进行交互协商以获取运行 作业所需的资源。 RM 的调度器根据集群的资源情况和配置的调度策略如容量调度、公平调度等将资源 分配给 MRAppMaster。任务执行 MRAppMaster 将获取到的资源分配给各个 NM。 NM 上的 Executor执行器根据分配到的资源和作业的具体任务开始执行 Map Task 或 Reduce Task。作业监控和完成 RM 的应用程序管理器负责监控 MRAppMaster 的运行状态并在必要时进行重启。 当所有任务执行完毕后MRAppMaster 通知 RM 作业已完成。 Client 通过 RM 获取作业完成状态并获取结果。MapperReducerMaster词频统计代码Mapperpackage edu.mapper; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class WordCountMapper extends MapperLongWritable, Text, Text, IntWritable { Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //获取读取到的每一行数据 String line value.toString(); //通过空格进行分割 String[] words line.split( ); //将初步处理的结果进行输出 for (String word: words) { context.write(new Text(word), new IntWritable(1)); } } }Reducerpackage edu.reducer; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class WordCountReducer extends ReducerText, IntWritable, Text,IntWritable { protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { Integer count 0; for(IntWritable value: values){ //将每一个分组中的数据进行累加计数 count value.get(); } //将最终结果进行输出 context.write(key, new IntWritable(count)); } }Masterpackage edu; import edu.mapper.WordCountMapper; import edu.reducer.WordCountReducer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class WordCountMaster { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { //初始化配置 Configuration conf new Configuration(); conf.set(fs.defaultFS, hdfs://virtual-machine-name:9000); //初始化job参数指定job名称 Job job Job.getInstance(conf,WordCount); //设置运行job的类 job.setJarByClass(WordCountMaster.class); //设置Mapper类 job.setMapperClass(WordCountMapper.class); //设置Reducer类 job.setReducerClass(WordCountReducer.class); //设置Map的输出数据类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); //设置Reducer的输出数据类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); //设置输入的路径 //FileInputFormat.setInputPaths(job, new Path(Hadoop_Urlinput/wordCount)); //设置输出的路径 //FileOutputFormat.setOutputPath(job, new Path(Hadoop_Urloutput/wordCount/1)); //指定该mapreduce程序数据的输入和输出路径 Path inputPath new Path(/wordcount/input); Path outputPath new Path(/wordcount/output); FileSystem fs FileSystem.get(conf); if(fs.exists(outputPath)){ fs.delete(outputPath, true); } FileInputFormat.setInputPaths(job, inputPath); FileOutputFormat.setOutputPath(job, outputPath); //提交job boolean result job.waitForCompletion(true); //执行成功后进行后续操作 if (result) { System.out.println(Congratulations!); } } }计算第四列每个元素出现的个数a,b,c,db,b,f,ea,a,c,fmapperimport java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class LogMapper extends MapperLongWritable, Text, Text, LongWritable { Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); String[] columns line.split(,); if (columns.length 4) { context.write(new Text(columns[3]), new LongWritable(1)); } } }reducerimport java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class LogReducer extends ReducerText, LongWritable, Text, LongWritable { Override protected void reduce(Text key, IterableLongWritable values, Context context) throws IOException, InterruptedException { long count 0; for (LongWritable value : values) { count value.get(); } context.write(key, new LongWritable(count)); } }mainimport org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class LogAnalysis { public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, Log Analysis); job.setJarByClass(LogAnalysis.class); job.setMapperClass(LogMapper.class); job.setReducerClass(LogReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
