一脸懵逼学习MapReduce的原理和编程（Map局部处理，Reduce汇总）和MapReduce几种运行方式

1：MapReduce的概述：

　　（1）：MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题.
　　（2）：MapReduce由两个阶段组成：Map和Reduce，用户只需要实现map()和reduce()两个函数，即可实现分布式计算，非常简单。
　　（3）：这两个函数的形参是key、value对，表示函数的输入信息。

2：MapReduce执行步骤：

　　（1）： map任务处理

　　　　（a）:读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。
　　　　（b）:写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
　　（2）reduce任务处理

　　　　（a）在reduce之前，有一个shuffle的过程对多个map任务的输出进行合并、排序。
　　　　（b）写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
　　　　（c）把reduce的输出保存到文件中。
　　　　　　 例子：实现WordCountApp
3：map、reduce键值对格式：

 4：MapReduce流程：
　　（1）代码编写
　　（2）作业配置
　　（3）提交作业
　　（4）初始化作业
　　（5）分配任务
　　（6）执行任务
　　（7）更新任务和状态
　　（8）完成作业


5：MapReduce介绍及wordcount和wordcount的编写和提交集群运行的案例：

WcMap类进行单词的局部处理：

 1 package com.mapreduce;
 2
 3
 4 import java.io.IOException;
 5
 6 import org.apache.commons.lang.StringUtils;
 7 import org.apache.hadoop.io.LongWritable;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Mapper;
10
11 /***
12  *
13  * @author Administrator
14  * 1：4个泛型中，前两个是指定mapper输入数据的类型，KEYIN是输入的key的类型，VALUEIN是输入的value的值
15  *       KEYOUT是输入的key的类型，VALUEOUT是输入的value的值
16  * 2：map和reduce的数据输入和输出都是以key-value的形式封装的。
17  * 3：默认情况下，框架传递给我们的mapper的输入数据中，key是要处理的文本中一行的起始偏移量，这一行的内容作为value
18  * 4：key-value数据是在网络中进行传递，节点和节点之间互相传递，在网络之间传输就需要序列化，但是jdk自己的序列化很冗余
19  *    所以使用hadoop自己封装的数据类型，而不要使用jdk自己封装的数据类型；
20  *    Long--->LongWritable
21  *    String--->Text
22  */
23 public class WcMap extends Mapper<LongWritable, Text, Text, LongWritable>{
24
25     //重写map这个方法
26     //mapreduce框架每读一行数据就调用一次该方法
27     @Override
28     protected void map(LongWritable key, Text value, Context context)
29             throws IOException, InterruptedException {
30         //具体业务逻辑就写在这个方法体中，而且我们业务要处理的数据已经被框架传递进来，在方法的参数中key-value
31         //key是这一行数据的起始偏移量，value是这一行的文本内容
32
33         //1:切分单词,首先拿到单词value的值,转化为String类型的
34         String str = value.toString();
35         //2:切分单词,空格隔开,返回切分开的单词
36         String[] words = StringUtils.split(str," ");
37         //3:遍历这个单词数组，输出为key-value的格式，将单词发送给reduce
38         for(String word : words){
39             //输出的key是Text类型的，value是LongWritable类型的
40             context.write(new Text(word), new LongWritable(1));
41         }
42
43
44     }
45 }

WcReduce进行单词的计数处理：

 1 package com.mapreduce;
 2
 3 import java.io.IOException;
 4
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Reducer;
 8
 9 /***
10  *
11  * @author Administrator
12  * 1:reduce的四个参数，第一个key-value是map的输出作为reduce的输入,第二个key-value是输出单词和次数，所以
13  *      是Text,LongWritable的格式；
14  */
15 public class WcReduce extends Reducer<Text, LongWritable, Text, LongWritable>{
16
17     //继承Reducer之后重写reduce方法
18     //第一个参数是key，第二个参数是集合。
19     //框架在map处理完成之后，将所有key-value对缓存起来，进行分组，然后传递一个组<key,valus{}>，调用一次reduce方法
20     //<hello,{1,1,1,1,1,1.....}>
21     @Override
22     protected void reduce(Text key, Iterable<LongWritable> values,Context context)
23             throws IOException, InterruptedException {
24         //将values进行累加操作，进行计数
25         long count = 0;
26         //遍历value的list，进行累加求和
27         for(LongWritable value : values){
28
29             count += value.get();
30         }
31
32         //输出这一个单词的统计结果
33         //输出放到hdfs的某一个目录上面，输入也是在hdfs的某一个目录
34         context.write(key, new LongWritable(count));
35     }
36
37
38 }

WcRunner用来描述一个特定的作业

 1 package com.mapreduce;
 2
 3 import java.io.IOException;
 4
 5 import org.apache.hadoop.conf.Configuration;
 6 import org.apache.hadoop.fs.Path;
 7 import org.apache.hadoop.io.LongWritable;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Job;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12
13
14 /***
15  * 1:用来描述一个特定的作业
16  *       比如,该作业使用哪个类作为逻辑处理中的map,那个作为reduce
17  * 2:还可以指定该作业要处理的数据所在的路径
18  *        还可以指定改作业输出的结果放到哪个路径
19  * @author Administrator
20  *
21  */
22 public class WcRunner {
23
24     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
25         //创建配置文件
26         Configuration conf = new Configuration();
27         //获取一个作业
28         Job job = Job.getInstance(conf);
29
30         //设置整个job所用的那些类在哪个jar包
31         job.setJarByClass(WcRunner.class);
32
33         //本job使用的mapper和reducer的类
34         job.setMapperClass(WcMap.class);
35         job.setReducerClass(WcReduce.class);
36
37         //指定reduce的输出数据key-value类型
38         job.setOutputKeyClass(Text.class);
39         job.setOutputValueClass(LongWritable.class);
40
41
42         //指定mapper的输出数据key-value类型
43         job.setMapOutputKeyClass(Text.class);
44         job.setMapOutputValueClass(LongWritable.class);
45
46         //指定要处理的输入数据存放路径
47         FileInputFormat.setInputPaths(job, new Path("hdfs://master:9000/wc/srcdata"));
48
49         //指定处理结果的输出数据存放路径
50         FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/wc/output"));
51
52         //将job提交给集群运行
53         job.waitForCompletion(true);
54     }
55
56 }

书写好上面的三个类以后打成jar包上传到虚拟机上面进行运行：

然后启动你的hadoop集群：start-dfs.sh和start-yarn.sh启动集群；然后将jar分发到节点上面进行运行；

之前先造一些数据，如下所示：

内容自己随便搞吧：

 然后上传到hadoop集群上面，首选创建目录，存放测试数据，将数据上传到创建的目录即可；但是输出目录不需要手动创建，会自动创建，自己创建会报错：

然后将jar分发到节点上面进行运行；命令格式如hadoop    jar   自己的jar包   主类的路径

 正常性运行完过后可以查看一下运行的效果：

6：MapReduce的本地模式运行如下所示（本地运行需要修改输入数据存放路径和输出数据存放路径）：

 1 package com.mapreduce;
 2
 3 import java.io.IOException;
 4
 5 import org.apache.hadoop.conf.Configuration;
 6 import org.apache.hadoop.fs.Path;
 7 import org.apache.hadoop.io.LongWritable;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Job;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12
13
14 /***
15  * 1:用来描述一个特定的作业
16  *       比如,该作业使用哪个类作为逻辑处理中的map,那个作为reduce
17  * 2:还可以指定该作业要处理的数据所在的路径
18  *        还可以指定改作业输出的结果放到哪个路径
19  * @author Administrator
20  *
21  */
22 public class WcRunner {
23
24     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
25         //创建配置文件
26         Configuration conf = new Configuration();
27         //获取一个作业
28         Job job = Job.getInstance(conf);
29
30         //设置整个job所用的那些类在哪个jar包
31         job.setJarByClass(WcRunner.class);
32
33         //本job使用的mapper和reducer的类
34         job.setMapperClass(WcMap.class);
35         job.setReducerClass(WcReduce.class);
36
37         //指定reduce的输出数据key-value类型
38         job.setOutputKeyClass(Text.class);
39         job.setOutputValueClass(LongWritable.class);
40
41
42         //指定mapper的输出数据key-value类型
43         job.setMapOutputKeyClass(Text.class);
44         job.setMapOutputValueClass(LongWritable.class);
45
46         //指定要处理的输入数据存放路径
47         //FileInputFormat.setInputPaths(job, new Path("hdfs://master:9000/wc/srcdata/"));
48         FileInputFormat.setInputPaths(job, new Path("d:/wc/srcdata/"));
49
50
51         //指定处理结果的输出数据存放路径
52         //FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9000/wc/output/"));
53         FileOutputFormat.setOutputPath(job, new Path("d:/wc/output/"));
54
55
56         //将job提交给集群运行
57         job.waitForCompletion(true);
58     }
59
60 }

然后去自己定义的盘里面创建文件夹即可：

然后直接运行出现下面的错误：

log4j:WARN No appenders could be found for logger (org.apache.hadoop.metrics2.lib.MutableMetricsFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Exception in thread "main" java.io.IOException: Cannot initialize Cluster. Please check your configuration for mapreduce.framework.name and the correspond server addresses.
    at org.apache.hadoop.mapreduce.Cluster.initialize(Cluster.java:120)
    at org.apache.hadoop.mapreduce.Cluster.<init>(Cluster.java:82)
    at org.apache.hadoop.mapreduce.Cluster.<init>(Cluster.java:75)
    at org.apache.hadoop.mapreduce.Job$9.run(Job.java:1255)
    at org.apache.hadoop.mapreduce.Job$9.run(Job.java:1251)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1556)
    at org.apache.hadoop.mapreduce.Job.connect(Job.java:1250)
    at org.apache.hadoop.mapreduce.Job.submit(Job.java:1279)
    at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:1303)
    at com.mapreduce.WcRunner.main(WcRunner.java:57)

解决办法：

缺少Jar包：hadoop-mapreduce-client-common-2.2.0.jar


 好吧，最后还是没有实现在本地运行此运行，先在这里记一下吧。下面这个错搞不定，先做下笔记吧；

log4j:WARN No appenders could be found for logger (org.apache.hadoop.metrics2.lib.MutableMetricsFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Exception
in thread "main" java.lang.IllegalArgumentException: Pathname
/c:/wc/output from hdfs://master:9000/c:/wc/output is not a valid DFS
filename.
    at org.apache.hadoop.hdfs.DistributedFileSystem.getPathName(DistributedFileSystem.java:194)
    at org.apache.hadoop.hdfs.DistributedFileSystem.access$000(DistributedFileSystem.java:102)
    at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:1124)
    at org.apache.hadoop.hdfs.DistributedFileSystem$17.doCall(DistributedFileSystem.java:1120)
    at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
    at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1120)
    at org.apache.hadoop.fs.FileSystem.exists(FileSystem.java:1398)
    at org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.checkOutputSpecs(FileOutputFormat.java:145)
    at org.apache.hadoop.mapreduce.JobSubmitter.checkSpecs(JobSubmitter.java:458)
    at org.apache.hadoop.mapreduce.JobSubmitter.submitJobInternal(JobSubmitter.java:343)
    at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1285)
    at org.apache.hadoop.mapreduce.Job$10.run(Job.java:1282)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:415)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1556)
    at org.apache.hadoop.mapreduce.Job.submit(Job.java:1282)
    at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:1303)
    at com.mapreduce.WcRunner.main(WcRunner.java:57)


7：MapReduce程序的几种提交运行模式：

本地模型运行
1：在windows的eclipse里面直接运行main方法，就会将job提交给本地执行器localjobrunner执行
      ----输入输出数据可以放在本地路径下（c:/wc/srcdata/）
      ----输入输出数据也可以放在hdfs中(hdfs://master:9000/wc/srcdata)
2：在linux的eclipse里面直接运行main方法，但是不要添加yarn相关的配置，也会提交给localjobrunner执行
      ----输入输出数据可以放在本地路径下（/home/hadoop/wc/srcdata/）
      ----输入输出数据也可以放在hdfs中(hdfs://master:9000/wc/srcdata)  
     
集群模式运行
1：将工程打成jar包，上传到服务器，然后用hadoop命令提交  hadoop jar wc.jar cn.itcast.hadoop.mr.wordcount.WCRunner
2：在linux的eclipse中直接运行main方法，也可以提交到集群中去运行，但是，必须采取以下措施：
      ----在工程src目录下加入 mapred-site.xml  和  yarn-site.xml
      ----将工程打成jar包(wc.jar)，同时在main方法中添加一个conf的配置参数　conf.set("mapreduce.job.jar","wc.jar");           

3：在windows的eclipse中直接运行main方法，也可以提交给集群中运行，但是因为平台不兼容，需要做很多的设置修改
        ----要在windows中存放一份hadoop的安装包（解压好的）
        ----要将其中的lib和bin目录替换成根据你的windows版本重新编译出的文件
        ----再要配置系统环境变量 HADOOP_HOME  和 PATH
        ----修改YarnRunner这个类的源码 

 

时间： 2024-08-01 13:17:53

一脸懵逼学习MapReduce的原理和编程（Map局部处理，Reduce汇总）和MapReduce几种运行方式的相关文章

一脸懵逼学习Hadoop中的序列化机制——流量求和统计MapReduce的程序开发案例——流量求和统计排序

一:序列化概念 序列化(Serialization)是指把结构化对象转化为字节流.反序列化(Deserialization)是序列化的逆过程.即把字节流转回结构化对象.Java序列化(java.io.Serializable) 二:Hadoop序列化的特点 (1):序列化格式特点: 紧凑:高效使用存储空间. 快速:读写数据的额外开销小. 可扩展:可透明地读取老格式的数据. 互操作:支持多语言的交互. (2):Hadoop的序列化格式:Writable接口 三:Hadoop序列化的作用: (1):

一脸懵逼学习Hive的使用以及常用语法（Hive语法即Hql语法）

Hive官网(HQL)语法手册(英文版):https://cwiki.apache.org/confluence/display/Hive/LanguageManual  Hive的数据存储 1.Hive中所有的数据都存储在 HDFS 中,没有专门的数据存储格式(可支持Text,SequenceFile,ParquetFile,RCFILE等) 2.只需要在创建表的时候告诉 Hive 数据中的列分隔符和行分隔符,Hive 就可以解析数据. 3.Hive 中包含以下数据模型:DB.Table,Ex

一脸懵逼学习基于CentOs的Hadoop集群安装与配置（三台机器跑集群）

1:Hadoop分布式计算平台是由Apache软件基金会开发的一个开源分布式计算平台.以Hadoop分布式文件系统(HDFS)和MapReduce(Google MapReduce的开源实现)为核心的Hadoop为用户提供了系统底层细节透明的分布式基础架构.  注意:HADOOP的核心组件有: 1)HDFS(分布式文件系统) 2)YARN(运算资源调度系统) 3)MAPREDUCE(分布式运算编程框架)       Hadoop 中的分布式文件系统 HDFS 由一个管理结点 ( NameNode

一脸懵逼学习Hadoop中的MapReduce程序中自定义分组的实现

1:首先搞好实体类对象: write 是把每个对象序列化到输出流,readFields是把输入流字节反序列化,实现WritableComparable,Java值对象的比较:一般需要重写toString(),hashCode(),equals()方法 1 package com.areapartition; 2 3 import java.io.DataInput; 4 import java.io.DataOutput; 5 import java.io.IOException; 6 7 im

一脸懵逼学习Hive的安装（将sql语句翻译成MapReduce程序的一个工具）

Hive只在一个节点上安装即可: 1.上传tar包:这个上传就不贴图了,贴一下上传后的,看一下虚拟机吧: 2.解压操作: [root@slaver3 hadoop]# tar -zxvf hive-0.12.0.tar.gz 解压后贴一下图:  3:解压缩以后启动一下hive:  4:开始操作sql: 好吧,开始没有启动集群,输入mysql创建数据库命令,直接不屌我,我也是苦苦等待啊: 5:启动我的集群,如下所示,这里最后帖一遍部署以后集群关了,重新开启集群的步骤,不能按照部署集群的时候进行格式

一脸懵逼学习Hive的元数据库Mysql方式安装配置

1:要想学习Hive必须将Hadoop启动起来,因为Hive本身没有自己的数据管理功能,全是依赖外部系统,包括分析也是依赖MapReduce: 2:七个节点跑HA集群模式的: 第一步:必须先将Zookeeper启动起来(HA里面好多组件都依赖Zookeeper): 切换目录,启动Zookeeper(master节点,slaver1节点,slaver2节点):./zkServer.sh start 第二步:启动HDFS(千万不要格式化了,不然肯定报错给你): 直接在slaver3节点启动:star

一脸懵逼学习Hadoop分布式集群HA模式部署（七台机器跑集群）

1)集群规划:主机名        IP      安装的软件                     运行的进程master    192.168.199.130   jdk.hadoop                     NameNode.DFSZKFailoverController(zkfc)slaver1    192.168.199.131    jdk.hadoop                       NameNode.DFSZKFailoverController(

一脸懵逼学习Hive（数据仓库基础构架）

Hive是什么?其体系结构简介*Hive的安装与管理*HiveQL数据类型,表以及表的操作*HiveQL查询数据***Hive的Java客户端** Hive的自定义函数UDF* 1:什么是Hive(一): (1)Hive 是建立在 Hadoop  上的数据仓库基础构架.它提供了一系列的工具,可以用来进行数据提取转化加载(ETL ),这是一种可以存储.查询和分析存储在 Hadoop  中的大规模数据的机制.Hive 定义了简单的类 SQL  查询语言,称为 QL ,它允许熟悉 SQL  的用户查询

一脸懵逼学习Storm---（一个开源的分布式实时计算系统）

Storm的官方网址:http://storm.apache.org/index.html 1:什么是Storm? Storm是一个开源的分布式实时计算系统,可以简单.可靠的处理大量的数据流.被称作"实时的hadoop".Storm有很多使用场景:如实时分析,在线机器学习,持续计算, 分布式RPC,ETL等等.Storm支持水平扩展,具有高容错性,保证每个消息都会得到处理,而且处理速度很快(在一个小集群中,每个结点每秒可以处理 数以百万计的消息).Storm的部署和运维都很便捷,而且更