MapReduce(一)之基本概念及 *** 作

MapReduce(一)之基本概念及 *** 作 MapReduce的概念及特点

MapReduce是Hadoop体系下的一种计算模型(计算框架|编程框架),主要是用来对存储在HDFS上的数据进行统计及分析的。

MapReduce特点

• 易于编程：只需要使用Hadoop接口进行编程，即可实现多台计算机分布式计算和分布式存储。
• 高扩展性：存储空间不足或者计算能力不足，则可以添加计算机完成。
• 容错性高：如果某个节点宕机，Hadoop可以自动切换讲计算任务转移到其他节点上完成，不会影响计算结果。
• 应用场景：PB级别以上海量数据的离线处理，无法实时处理和流失动态处理。(每日)

MapReduce的核心思想

分而治之：大任务拆分小任务。

MapReduce计算

• Job(一个大型任务)[Application]

  一组MapReduce又统称为一个Job作业

• Map(拆分后的小任务)

  局部计算

• Reduce(整合任务)

  对局部计算结果进行汇总计算

yarn框架

作用(施工队)：资源调度、任务监控。主要用来整合Hadoop集群中的资源(CPU、内存)进行统一调度，同时监控任务的执行情况。

​ 总结: 联合多个服务器节点的硬件，共同完成一个计算。突破单机服务器的计算能力。

组成部分：

• ResourceManager(老板)

  集群计算资源的管理器，也是yarn架构中的主节点。
  功能：

    	1. 监控集群资源
    	2. 为计算分配资源
  

• NodeManager(员工)

  yarn集群计算资源的提供者，也是yarn架构中的从节点。
  功能：

  1. 真正执行计算任务的节点。
  2. 监控本节点的资源情况(CPU、内存、网络、硬盘)，并通过心跳向ResourceManager汇报。

Yarn伪分布式搭建

# 准备单机的HDFS架构
验证：jps

[root@hadoop10 ~]# jps
2224 Jps
2113 SecondaryNameNode
1910 DataNode
1806 NameNode

关闭掉hdfs
	stop-dfs.sh
# 初始化配置文件

1. mapred-site.xml
	
	
        mapreduce.framework.name
        yarn
    
2. yarn-site.xml
	
        yarn.nodemanager.aux-services
        mapreduce_shuffle
    
	
    
        yarn.resourcemanager.hostname
        Hadoop
    

# 启动yarn集群
1. 启动HDFS集群
	start-dfs.sh
2. 启动yarn集群
	start-yarn.sh

# 验证
1. jps
[root@hadoop11 ~]# jps
    6160 DataNode
    6513 ResourceManager
    6614 NodeManager
    6056 NameNode
    6349 SecondaryNameNode
    6831 Jps
2. 访问yarn的资源调度器web网页。
	http://ip:8088

MapReduce数据流转机制

1. InputFormat(MapReduce自动处理)
	将block文件转化成split切片，其中每条数据由key-value组成。
	key是数据偏移量，value是每条数据。
2. Map(程序员编码)
	将split逐条输入给map，由map负责，对每条数据进行处理，转化为keyOut-valueOut。
3. Shuffle(MapReduce的默认处理器)
	对map输出的每条数据的key-value进行排序，分组。
4. Reduce(程序员编码)
	对Shuffle分组后的数据的key-value进行处理，转化为新的key-value。
5. OutputFormat
	将reduce产生的数据，存储HDFS文件系统中。

MapReduce编码

# 编写map程序

public class NameCountMap extends Mapper {
    
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //局部计算
        String[] names = value.toString().split(" ");
        for (String name : names) {
            context.write(new Text(name),new IntWritable(1));
        }
    }
}

# 编写reduce程序

public class NameCountReduce extends Reducer {
    
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            int i = value.get();
            sum+=i;
        }
        context.write(key,new IntWritable(sum));
    }
}

# 编写job程序

		//0. 初始化HDFS配置文件
		Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.242.10:9000");
        //1. 创建一个Job
        Job job = Job.getInstance(conf);
        job.setJarByClass(MR1.class);
        //2. 设置map  key value mapclass InputFormat
        job.setMapperClass(NameCountMap.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        TextInputFormat.addInputPath(job,new Path("/hdfs/mr/wordcount.log"));
        //3. 设置reduce  key value reduce class TextOutputFormat
        job.setReducerClass(NameCountReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        TextOutputFormat.setOutputPath(job,new Path("/hdfs/mrout"));//该目录是mapreduce生成的，不能自己创建，并且不能存在。
        //4. 启动Job
        boolean b = job.waitForCompletion(true);
        System.out.println(b);

# 本地直接运行
	使用本地的方式提交任务，需要HDFS开启写入文件的权限。
	hdfs dfs -chmod -R 777 /hdfs

MapReduce任务提交 打包上传任务

# 打包
# 1. 设置maven的打包的环境

    
    UTF-8


    
    word_count
    
    
        
            org.apache.maven.plugins
            maven-jar-plugin
            2.4
            
                
                    
                    
                        com.aries.mapreduce_day01_01.job.WordCountJob
                    
                
            
        
    

# 2. 执行打包
	在当前项目所在的目录下执行如下命令
	> mvn package
# 3. 上传jar包到Hadoop的ResourceManager所在的机器
# 4. 执行程序
	> yarn jar mr1.jar

maven自动化部署插件wagon

# 1. 配置maven远程提交插件
1. 添加maven的ssh扩展
2. 添加maven的远程拷贝插件wagon(货车)

    
        org.apache.maven.wagon
        wagon-ssh
        2.8
    

    org.codehaus.mojo
    wagon-maven-plugin
    1.0
    
        
        target/${project.build.finalName}.jar
        
        scp://用户名:密码@ip:/opt/app
    

3. 添加远程执行命令，和参数。

# 清空
	mvn clean
# 打包本地jar
	mvn package
# 远程上传jar
	mvn wagon:upload-single

4. 远程执行命令
	# 如下内容在wagon的插件中添加

    org.codehaus.mojo
    wagon-maven-plugin
    1.0
    
        
        target/${project.build.finalName}.jar
        
        scp://root:admins@192.168.242.10:/opt/app
        
        
            
            nohup /opt/installs/hadoop2.9.2/bin/yarn jar /opt/app/${project.build.finalName}.jar > /opt/logs/mr.out 2>&1 &
        
        
        true
    

nohup命令简介

nohupno hang up 命令执行后，要持续一段时间，不要挂断。&(结尾)最后的&表示nohup命令的结尾。2>&1错误和输出都重定向写入到前面的文件中。命令参数含义

# maven和wagon 命令汇总
 `注意，mvn命令需要在项目的pom.xml所在的目录执行`
1. 清空
	mvn clean
2. 打包
	mvn package
3. 上传
	mvn wagon:upload-single 
4. 执行
	mvn wagon:sshexec

配置MapReduce的历史日志服务器

Hadoop自带了一个历史服务器，可以通过历史服务器查看已经运行完的MapReduce作业记录。

比如用了多少个Map、用了多少个Reduce、作业提交时间、作业启动时间、作业完成时间等信息。

默认未启动

# 1. 配置mapred-site.xml，指定历史日志服务器的地址

   mapreduce.jobhistory.address
   hadoop10:10020



   mapreduce.jobhistory.webapp.address
   hadoop10:19888

# 2. 配置yarn-site.xml,指定开启日志聚合和日志保留时间


  yarn.log-aggregation-enable
  true


 
  yarn.log-aggregation.retain-seconds
  604800

# 3. 启动历史日志服务器
1. 重启yarn
	[root@hadoop10 ~]# stop-yarn.sh
	[root@hadoop10 ~]# start-yarn.sh
2. 启动
	[root@hadoop10 ~]# mr-jobhistory-daemon.sh start historyserver
	如果需要关闭执行如下命令
	[root@hadoop10 ~]# mr-jobhistory-daemon.sh stop historyserver

# 4. 查看日志
	1. 访问http://ip:8088(访问yarn集群，看到执行过的job信息)
	2. 点击"Applications"找到刚才执行的job的"history"
	3. 点击logs