用于大规模数据集(大于1TB)的并行运算的MapReduce是怎么实现的?

本文主要是介绍用于大规模数据集(大于1TB)的并行运算的MapReduce是怎么实现的?,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

MapReduce 是一种编程模型,用于处理和生成大数据集。MapReduce 分为两个阶段:Map 阶段和 Reduce 阶段。

  1. Map 阶段:在这个阶段,输入数据被拆分成不同的数据块,这些数据块被分发到各个 Map 任务上。每个 Map 任务对输入的数据块进行处理并输出成一组键值对。具体的处理方式取决于编写 Map 任务的方式,这将由程序员决定。

  2. Shuffle 阶段:在 Map 阶段之后,系统会根据 Map 阶段输出的键值对进行排序和分区。这个过程是系统自动完成的,不需要程序员编写代码。

  3. Reduce 阶段:在这个阶段,一组键值对会送到同一个 Reduce 任务上。Reduce 任务会对这些键值对按照键来进行合并,并对每个键的所有值进行处理,输出处理结果。具体的处理方式也是由程序员编写的 Reduce 任务来决定。

通过 Map 和 Reduce 两个步骤,MapReduce 能够将大规模的数据计算工作分解成小规模的计算任务,这些小规模的计算任务可以在不同的计算节点上并行处理。这使得 MapReduce 可以处理非常大规模的数据。

例如,一个简单的MapReduce程序可能会将文本文件分成单词(map),然后计算每个单词的出现次数(reduce)。在这种情况下,Map任务将文本文件切分为单词,并输出每个单词的键值对(键是单词,值是1),然后Reduce任务将这些键值对按照键(单词)合并,并计算每个键(单词)的值(出现次数)的总和。

在真实的环境中,这个模型通常由分布式计算框架(如 Hadoop)来实现,框架会负责任务的调度、数据的分布和容错等工作,使得程序员可以专注于编写处理数据的 Map 和 Reduce 函数。

上demo:

// 导入 IOException 类,用于处理输入/输出错误
import java.io.IOException;// 导入 StringTokenizer 类,用于解析字符串
import java.util.StringTokenizer;// 导入 Hadoop 提供的 Configuration 类,用于 Hadoop 配置设置
import org.apache.hadoop.conf.Configuration;// 导入 Hadoop 提供的 Path 类,处理 Hadoop 文件系统路径
import org.apache.hadoop.fs.Path;// 导入 Hadoop 提供的 IntWritable 和 Text 类,数据类型转换
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;// 导入 Hadoop MapReduce Job 类,用于定义和提交 MapReduce 任务
import org.apache.hadoop.mapreduce.Job;// 从 Hadoop MapReduce 库导入 Mapper 和 Reducer 类
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;// 导入 Hadoop IO 类库,用于 MapReduce 输入输出
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;// 定义 WordCount 类
public class WordCount {// 定义公开的静态 TokenizerMapper 类,继承 Hadoop 的 Mapper 类public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{// 定义一个值为 1 的整型变量,作为每个单词计数的值private final static IntWritable one = new IntWritable(1);// 定义一个 Text 类型的变量,存储每个即将被计数的单词private Text word = new Text();// 重写 map 方法,这个方法会被 MapReduce 框架调用,每读入一行数据调用一次public void map(Object key, Text value, Context context) throws IOException, InterruptedException {// 使用 StringTokenizer 将每行的字符串拆解为单个的单词StringTokenizer itr = new StringTokenizer(value.toString());// 循环单词字符串while (itr.hasMoreTokens()) {// 为 word 对象设值word.set(itr.nextToken());// 将每个单词及其数量(默认为 1)输出context.write(word, one);}}}// 定义公开的静态 IntSumReducer 类,继承 Hadoop 的 Reducer 类public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> {// 定义 IntWritable 类型的对象,存储每个单词的计数总和private IntWritable result = new IntWritable();// 重写 reduce 方法,这个方法用于汇总每个单词的数量public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {// 定义 sum 变量,用于累加每个单词的数量int sum = 0;// 使用 for 循环,遍历当前单词的所有数量值,进行累加for (IntWritable val : values) {sum += val.get();}// 为结果设值result.set(sum);// 将每个单词及其最终计数总和写出context.write(key, result);}}// 定义主方法,运行 MapReduce 任务的入口public static void main(String[] args) throws Exception {// 创建 Hadoop 配置对象Configuration conf = new Configuration();// 创建 Job 实例,设置 Job 名称Job job = Job.getInstance(conf, "word count");// 设置 Jar 文件 这行代码的作用是设置该作业的 jar 文件。在 Hadoop 中执行一个 MapReduce 作业,需要将其打包为一个 jar 文件并上传到 Hadoop 集群中。这里的 WordCount.class 即是你的作业驱动类, Hadoop 将会通过这个类查找包含该类的 jar 文件,然后使用这个 jar 文件执行整个作业。这样设置可以确保作业在 Hadoop 集群中的每个节点上都可以正确执行。job.setJarByClass(WordCount.class);// 设置 Mapper 类 这行代码的作用是设置 MapReduce 作业的 Mapper 类。在 MapReduce 模型中,Mapper 是负责处理输入数据,将输入数据转换为一系列键值对的组件。TokenizerMapper.class 就是你写的处理数据的类,你可以在这个类中定义如何处理和映射输入数据。设置 Mapper 类是进行 MapReduce 作业的关键步骤之一。job.setMapperClass(TokenizerMapper.class);// 设置 Combiner 类  当一个特定的 Mapper 结束处理之后,它会输出很多记录,这些记录中可能有很多键是相同的。如果没有 Combiner,这些记录就会直接传递给 Reducer,这样可能会产生大量的网络 I/O,显著地影响性能。//而有了 Combiner,我们可以在每个 Mapper 所在的节点上先进行一次局部的归约操作,将相同键的值合并后再传输到 Reducer,这样可以大大减少需要传输的数据量,提高任务的运行性能。job.setCombinerClass(IntSumReducer.class);// 设置 Reducer 类及其输出键-值对的类型// 这行代码是设定了该 MapReduce 任务的 Reducer 类是 IntSumReducer。Reducer 类是对所有 Mapper 的输出进行汇总的部分,每一个 MapReduce 任务都需要设定一个 Reducer 类。这里的 IntSumReducer 类定义了如何对数据进行汇总操作。job.setReducerClass(IntSumReducer.class); //这行代码是设定了该 MapReduce 任务输出数据的键的类型是 Text。在 MapReduce 中,数据是以键值对的形式存储和传输的。这行代码指明了输出键的数据类型,确保处理结果的格式正确。job.setOutputKeyClass(Text.class);//这行代码是设定了该 MapReduce 任务输出数据的值的类型是 IntWritable。这是对job.setOutputKeyClass(Text.class)的补充,指明了输出值的数据类型。job.setOutputValueClass(IntWritable.class);// 设置输入数据的路径,取自程序参数FileInputFormat.addInputPath(job, new Path(args[0]));// 设置输出数据的路径,取自程序参数FileOutputFormat.setOutputPath(job, new Path(args[1]));// 等待任务完成后退出程序 这行代码是 MapReduce 任务的最后一步,作用是让主线程等待这个 MapReduce 任务完成。//在这段代码中,waitForCompletion方法将主线程挂起,直到 MapReduce 任务完成为止。这里的参数 true 表示会在控制台上显示这个任务的进度,也就是说你运行这个程序的时候,会在控制台上看到任务的完成进度。//这段代码的含义就是:如果任务成功完成,那么程序正常退出;如果任务执行过程中出现了错误,那么程序异常退出。System.exit(job.waitForCompletion(true) ? 0 : 1);}
}
/**
MapReduce 模型的并发性体现在其 Map 和 Reduce 操作的实现中,这两个操作对数据进行分布式处理。在以上代码片段中,没有一行明确地标示了并发操作,因为并发性质是 MapReduce 框架内部管理的,并非是在业务代码中直接表现出来的。
然而,job.setMapperClass(TokenizerMapper.class); 和 job.setReducerClass(IntSumReducer.class); 
这两行设置了 Mapper 类和 Reducer 类,这两类的实例在运行时会在多个节点上同时处理数据(一个节点处理输入数据的一部分),这便是 MapReduce 的并发性!!!
具体来说:
在 Mapper 阶段,输入的数据集会被划分成多个数据块,每一个数据块会被一个 Mapper 实例负责处理,这些 Mapper 实例是在集群的各个计算节点上同时运行的。
Reducer 阶段同样具有并发性,不同的 Reducer 实例会并发地处理 Mapper 输出的数据集中具有相同键的数据子集。
请注意,这些并发处理是由 Hadoop 框架自动管理的,并不需要在业务代码中明确地进行并发控制,这也是使用这种大数据处理框架的优点之一。
*/

这个程序把一个任务分为两个部分:

  1. Mapper (TokenizerMapper class):接受一行文本,拆分成单词,然后返回一系列的<单词,1>这样的键值对。
  2. Reducer (IntSumReducer class):对所有相同的单词计数总和,并返回<单词,总计数>这样的键值对。

执行这个程序,需要有一个 Hadoop 环境并且把待统计名词的文本文件放到应当的路径,并将结果输出的路径设置为期望的路径进行查看结果。

请注意,这些并发处理是由 Hadoop 框架自动管理的,并不需要在业务代码中明确地进行并发控制,这也是使用这种大数据处理框架的优点之一。

这篇关于用于大规模数据集(大于1TB)的并行运算的MapReduce是怎么实现的?的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/889036

相关文章

AI绘图怎么变现?想做点副业的小白必看!

在科技飞速发展的今天,AI绘图作为一种新兴技术,不仅改变了艺术创作的方式,也为创作者提供了多种变现途径。本文将详细探讨几种常见的AI绘图变现方式,帮助创作者更好地利用这一技术实现经济收益。 更多实操教程和AI绘画工具,可以扫描下方,免费获取 定制服务:个性化的创意商机 个性化定制 AI绘图技术能够根据用户需求生成个性化的头像、壁纸、插画等作品。例如,姓氏头像在电商平台上非常受欢迎,

大模型研发全揭秘:客服工单数据标注的完整攻略

在人工智能(AI)领域,数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者,掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中,工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注,不仅能够帮助提升客服自动化系统的智能化水平,还能优化客户服务流程,提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后,随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步,我们目前的解决方案是建立数据中间表。把需要检索的业务数据,统一放到一张M

关于数据埋点,你需要了解这些基本知识

产品汪每天都在和数据打交道,你知道数据来自哪里吗? 移动app端内的用户行为数据大多来自埋点,了解一些埋点知识,能和数据分析师、技术侃大山,参与到前期的数据采集,更重要是让最终的埋点数据能为我所用,否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式,可以区分为: 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理:自动程度高的,能解决通用统计,便于统一化管理,但个性化定

W外链微信推广短连接怎么做?

制作微信推广链接的难点分析 一、内容创作难度 制作微信推广链接时,首先需要创作有吸引力的内容。这不仅要求内容本身有趣、有价值,还要能够激起人们的分享欲望。对于许多企业和个人来说,尤其是那些缺乏创意和写作能力的人来说,这是制作微信推广链接的一大难点。 二、精准定位难度 微信用户群体庞大,不同用户的需求和兴趣各异。因此,制作推广链接时需要精准定位目标受众,以便更有效地吸引他们点击并分享链接

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

异构存储(冷热数据分离)

异构存储主要解决不同的数据,存储在不同类型的硬盘中,达到最佳性能的问题。 异构存储Shell操作 (1)查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies (2)为指定路径(数据存储目录)设置指定的存储策略 hdfs storagepolicies -setStoragePo

Hadoop集群数据均衡之磁盘间数据均衡

生产环境,由于硬盘空间不足,往往需要增加一块硬盘。刚加载的硬盘没有数据时,可以执行磁盘数据均衡命令。(Hadoop3.x新特性) plan后面带的节点的名字必须是已经存在的,并且是需要均衡的节点。 如果节点不存在,会报如下错误: 如果节点只有一个硬盘的话,不会创建均衡计划: (1)生成均衡计划 hdfs diskbalancer -plan hadoop102 (2)执行均衡计划 hd

hdu1043(八数码问题,广搜 + hash(实现状态压缩) )

利用康拓展开将一个排列映射成一个自然数,然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu

电脑桌面文件删除了怎么找回来?别急,快速恢复攻略在此

在日常使用电脑的过程中,我们经常会遇到这样的情况:一不小心,桌面上的某个重要文件被删除了。这时,大多数人可能会感到惊慌失措,不知所措。 其实,不必过于担心,因为有很多方法可以帮助我们找回被删除的桌面文件。下面,就让我们一起来了解一下这些恢复桌面文件的方法吧。 一、使用撤销操作 如果我们刚刚删除了桌面上的文件,并且还没有进行其他操作,那么可以尝试使用撤销操作来恢复文件。在键盘上同时按下“C