Hadoop-MapReduce

MapReduce 是一个框架，我们可以使用它编写应用程序，以可靠的方式在大型商用硬件集群上并行处理大量数据。

什么是MapReduce？

MapReduce是一种基于java的分布式计算的处理技术和程序模型。MapReduce算法包含两个重要的任务，即Map和Reduce。Map 获取一组数据并将其转换为另一组数据，其中各个元素被分解为元组（键/值对）。其次，reduce 任务，它将映射的输出作为输入，并将这些数据元组组合成较小的元组集合。正如MapReduce名称的顺序所暗示的，reduce任务总是在map作业之后执行。

MapReduce 的主要优点是可以轻松地在多个计算节点上扩展数据处理。在MapReduce模型下，数据处理原语称为映射器和缩减器。将数据处理应用程序分解为映射器和化简器有时并不简单。但是，一旦我们以 MapReduce 形式编写应用程序，将应用程序扩展为在集群中的数百、数千甚至数万台机器上运行就仅仅是配置更改。这种简单的可扩展性吸引了许多程序员使用MapReduce模型。

算法

一般来说，MapReduce 范例是基于将计算机发送到数据所在的位置！
MapReduce程序分三个阶段执行，即map阶段、shuffle阶段和reduce阶段。
- 地图阶段- 地图或映射器的工作是处理输入数据。输入数据通常以文件或目录的形式存储在Hadoop文件系统（HDFS）中。输入文件逐行传递给映射器函数。映射器处理数据并创建几个小数据块。
- 减少阶段- 该阶段是随机播放阶段和减少阶段的组合。减速器的工作是处理来自映射器的数据。处理后，它会产生一组新的输出，并将其存储在 HDFS 中。
在MapReduce 作业期间，Hadoop 将Map 和Reduce 任务发送到集群中适当的服务器。
该框架管理数据传递的所有细节，例如发布任务、验证任务完成以及在集群节点之间复制数据。
大部分计算发生在节点上，数据存储在本地磁盘上，从而减少了网络流量。
完成给定任务后，集群收集并减少数据以形成适当的结果，并将其发送回 Hadoop 服务器。

输入和输出（Java 视角）

MapReduce框架对<key, value>对进行操作，也就是说，框架将作业的输入视为一组<key, value>对，并产生一组<key, value>对作为作业的输出，可以想象为不同类型。

键和值类应该由框架以序列化的方式进行，因此需要实现 Writable 接口。此外，关键类必须实现 Writable-Comparable 接口，以方便框架进行排序。MapReduce 作业的输入和输出类型- （输入）<k1, v1> → map → <k2, v2> → reduce → <k3, v3>（输出）。

	输入	输出
地图	<k1，v1>	列表 (<k2, v2>)
减少	<k2，列表（v2）>	列表 (<k3, v3>)

术语

PayLoad - 应用程序实现Map和Reduce功能，并构成作业的核心。
Mapper - Mapper 将输入键/值对映射到一组中间键/值对。
NamedNode - 管理 Hadoop 分布式文件系统 (HDFS) 的节点。
DataNode - 在进行任何处理之前提前呈现数据的节点。
MasterNode - JobTracker 运行并接受来自客户端的作业请求的节点。
SlaveNode - Map 和Reduce 程序运行的节点。
JobTracker - 安排作业并跟踪分配给任务跟踪器的作业。
任务跟踪器- 跟踪任务并向 JobTracker 报告状态。
作业- 程序是跨数据集的映射器和减速器的执行。
任务- 在数据切片上执行映射器或减速器。
任务尝试- 尝试在 SlaveNode 上执行任务的特定实例。

示例场景

下面给出的是有关组织的电力消耗的数据。它包含了每月的用电量和历年的年平均值。

	扬	二月	三月	四月	可能	君	七月	八月	九月	十月	十一月	十二月	平均
1979年	23	23	2	43	24	25	26	26	26	26	25	26	25
1980年	26	27	28	28	28	30	31	31	31	30	30	30	29
1981年	31	32	32	32	33	34	35	36	36	34	34	34	34
1984年	39	38	39	39	39	41	42	43	40	39	38	38	40
1985年	38	39	39	39	39	41	41	41	00	40	39	39	45

如果将上述数据作为输入给出，我们必须编写应用程序来处理它并生成结果，例如查找最大使用年份、最小使用年份等。对于记录数量有限的程序员来说，这是一个轻松的选择。他们将简单地编写逻辑来产生所需的输出，并将数据传递给所编写的应用程序。

但是，想想代表特定州自成立以来所有大型工业的电力消耗的数据。

当我们编写应用程序来处理如此大的数据时，

他们将花费很多时间来执行。
当我们将数据从源移动到网络服务器等时，将会产生大量的网络流量。

为了解决这些问题，我们有了MapReduce框架。

输入数据

上述数据保存为sample.txt并作为输入给出。输入文件如下所示。

1979   23   23   2   43   24   25   26   26   26   26   25   26  25 
1980   26   27   28  28   28   30   31   31   31   30   30   30  29 
1981   31   32   32  32   33   34   35   36   36   34   34   34  34 
1984   39   38   39  39   39   41   42   43   40   39   38   38  40 
1985   38   39   39  39   39   41   41   41   00   40   39   39  45

示例程序

下面给出的是使用 MapReduce 框架对示例数据进行处理的程序。

package hadoop; 

import java.util.*; 

import java.io.IOException; 
import java.io.IOException; 

import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.conf.*; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapred.*; 
import org.apache.hadoop.util.*; 

public class ProcessUnits {
   //Mapper class 
   public static class E_EMapper extends MapReduceBase implements 
   Mapper<LongWritable ,/*Input key Type */ 
   Text,                /*Input value Type*/ 
   Text,                /*Output key Type*/ 
   IntWritable>        /*Output value Type*/ 
   {
      //Map function 
      public void map(LongWritable key, Text value, 
      OutputCollector<Text, IntWritable> output,   
      
      Reporter reporter) throws IOException { 
         String line = value.toString(); 
         String lasttoken = null; 
         StringTokenizer s = new StringTokenizer(line,"\t"); 
         String year = s.nextToken(); 
         
         while(s.hasMoreTokens()) {
            lasttoken = s.nextToken();
         }
         int avgprice = Integer.parseInt(lasttoken); 
         output.collect(new Text(year), new IntWritable(avgprice)); 
      } 
   }
   
   //Reducer class 
   public static class E_EReduce extends MapReduceBase implements Reducer< Text, IntWritable, Text, IntWritable > {
   
      //Reduce function 
      public void reduce( Text key, Iterator <IntWritable> values, 
      OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { 
         int maxavg = 30; 
         int val = Integer.MIN_VALUE; 
            
         while (values.hasNext()) { 
            if((val = values.next().get())>maxavg) { 
               output.collect(key, new IntWritable(val)); 
            } 
         }
      } 
   }

   //Main function 
   public static void main(String args[])throws Exception { 
      JobConf conf = new JobConf(ProcessUnits.class); 
      
      conf.setJobName("max_eletricityunits"); 
      conf.setOutputKeyClass(Text.class);
      conf.setOutputValueClass(IntWritable.class); 
      conf.setMapperClass(E_EMapper.class); 
      conf.setCombinerClass(E_EReduce.class); 
      conf.setReducerClass(E_EReduce.class); 
      conf.setInputFormat(TextInputFormat.class); 
      conf.setOutputFormat(TextOutputFormat.class); 
      
      FileInputFormat.setInputPaths(conf, new Path(args[0])); 
      FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
      
      JobClient.runJob(conf); 
   } 
}

将上述程序保存为ProcessUnits.java。下面解释程序的编译和执行。

过程单元程序的编译和执行

假设我们位于 Hadoop 用户的主目录中（例如 /home/hadoop）。

按照下面给出的步骤编译并执行上述程序。

步骤1

下面的命令是创建一个目录来存放编译好的java类。

$ mkdir units

第2步

下载Hadoop-core-1.2.1.jar，用于编译并执行MapReduce程序。访问以下链接mvnrepository.com下载该 jar。我们假设下载的文件夹是/home/hadoop/。

步骤3

以下命令用于编译ProcessUnits.java程序并为该程序创建 jar。

$ javac -classpath hadoop-core-1.2.1.jar -d units ProcessUnits.java 
$ jar -cvf units.jar -C units/ .

步骤4

以下命令用于在 HDFS 中创建输入目录。

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

步骤5

以下命令用于复制HDFS输入目录中名为sample.txt的输入文件。

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/sample.txt input_dir

步骤6

以下命令用于验证输入目录中的文件。

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

步骤7

以下命令用于通过从输入目录获取输入文件来运行 Eleunit_max 应用程序。

$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir

等待一段时间，直到文件被执行。执行后，如下所示，输出将包含输入split的数量、Map任务的数量、reducer任务的数量等。

INFO mapreduce.Job: Job job_1414748220717_0002 
completed successfully 
14/10/31 06:02:52 
INFO mapreduce.Job: Counters: 49 
   File System Counters 
 
FILE: Number of bytes read = 61 
FILE: Number of bytes written = 279400 
FILE: Number of read operations = 0 
FILE: Number of large read operations = 0   
FILE: Number of write operations = 0 
HDFS: Number of bytes read = 546 
HDFS: Number of bytes written = 40 
HDFS: Number of read operations = 9 
HDFS: Number of large read operations = 0 
HDFS: Number of write operations = 2 Job Counters 


   Launched map tasks = 2  
   Launched reduce tasks = 1 
   Data-local map tasks = 2  
   Total time spent by all maps in occupied slots (ms) = 146137 
   Total time spent by all reduces in occupied slots (ms) = 441   
   Total time spent by all map tasks (ms) = 14613 
   Total time spent by all reduce tasks (ms) = 44120 
   Total vcore-seconds taken by all map tasks = 146137 
   Total vcore-seconds taken by all reduce tasks = 44120 
   Total megabyte-seconds taken by all map tasks = 149644288 
   Total megabyte-seconds taken by all reduce tasks = 45178880 
   
Map-Reduce Framework 
 
   Map input records = 5  
   Map output records = 5   
   Map output bytes = 45  
   Map output materialized bytes = 67  
   Input split bytes = 208 
   Combine input records = 5  
   Combine output records = 5 
   Reduce input groups = 5  
   Reduce shuffle bytes = 6  
   Reduce input records = 5  
   Reduce output records = 5  
   Spilled Records = 10  
   Shuffled Maps  = 2  
   Failed Shuffles = 0  
   Merged Map outputs = 2  
   GC time elapsed (ms) = 948  
   CPU time spent (ms) = 5160  
   Physical memory (bytes) snapshot = 47749120  
   Virtual memory (bytes) snapshot = 2899349504  
   Total committed heap usage (bytes) = 277684224
     
File Output Format Counters 
 
   Bytes Written = 40

步骤8

以下命令用于验证输出文件夹中的结果文件。

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

步骤9

以下命令用于查看Part-00000文件中的输出。该文件由 HDFS 生成。

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

下面是 MapReduce 程序生成的输出。

1981    34 
1984    40 
1985    45

步骤10

以下命令用于将输出文件夹从 HDFS 复制到本地文件系统进行分析。

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000/bin/hadoop dfs get output_dir /home/hadoop

重要命令

所有 Hadoop 命令均由$HADOOP_HOME/bin/hadoop命令调用。运行不带任何参数的 Hadoop 脚本会打印所有命令的描述。

用法- hadoop [--config confdir] 命令

下表列出了可用选项及其说明。

先生。	选项和说明
1	名称节点格式格式化 DFS 文件系统。
2	辅助名称节点运行 DFS 辅助名称节点。
3	名称节点运行 DFS 名称节点。
4	数据节点运行 DFS 数据节点。
5	dfs管理员运行 DFS 管理客户端。
6	先生管理员运行 Map-Reduce 管理客户端。
7	FSCK 运行 DFS 文件系统检查实用程序。
8	FS 运行通用文件系统用户客户端。
9	平衡器运行集群平衡实用程序。
10	奥伊夫将离线 fsimage 查看器应用于 fsimage。
11	获取数据从 NameNode 获取委托令牌。
12	工作跟踪器运行 MapReduce 作业跟踪器节点。
13	管道运行管道作业。
14	任务跟踪器运行 MapReduce 任务 Tracker 节点。
15	历史服务器将作业历史记录服务器作为独立守护进程运行。
16	工作操作 MapReduce 作业。
17 号	队列获取有关 JobQueue 的信息。
18	版本打印版本。
19	罐子<罐子> 运行 jar 文件。
20	distcp <srcurl> <desturl> 递归复制文件或目录。
21	distcp2 <srcurl> <desturl> DistCp 版本 2。
22	*archive -archiveName NAME -p <父路径> <源> <目标>** 创建 hadoop 存档。
23	类路径打印获取 Hadoop jar 和所需库所需的类路径。
24	守护进程日志获取/设置每个守护进程的日志级别

如何与 MapReduce 作业交互

用法 - hadoop 作业 [GENERIC_OPTIONS]

以下是 Hadoop 作业中可用的通用选项。

先生。	GENERIC_OPTION 和说明
1	-提交<作业文件> 提交作业。
2	-status <作业 ID> 打印地图并减少完成百分比和所有作业计数器。
3	-counter <作业 ID> <组名称> <计数器名称> 打印计数器值。
4	-kill <工作 ID> 杀死工作。
5	-events <作业 ID> <来自事件#> <事件#> 打印 jobtracker 收到的给定范围内的事件详细信息。
6	-history [全部] <jobOutputDir> - 历史 <jobOutputDir> 打印作业详细信息、失败和终止的提示详细信息。可以通过指定 [all] 选项来查看有关作业的更多详细信息，例如成功的任务和为每个任务所做的任务尝试。
7	-列表[全部] 显示所有作业。-list 仅显示尚未完成的作业。
8	-kill-task <任务 ID> 杀死任务。被终止的任务不计入失败的尝试。
9	-失败任务 <任务 ID> 任务失败。失败的任务将根据失败的尝试进行计数。
10	-set-priority <作业 ID> <优先级> 更改作业的优先级。允许的优先级值为 VERY_HIGH、HIGH、NORMAL、LOW、VERY_LOW

查看作业状态

$ $HADOOP_HOME/bin/hadoop job -status <JOB-ID> 
e.g. 
$ $HADOOP_HOME/bin/hadoop job -status job_201310191043_0004

查看作业输出目录的历史记录

$ $HADOOP_HOME/bin/hadoop job -history <DIR-NAME> 
e.g. 
$ $HADOOP_HOME/bin/hadoop job -history /user/expert/output

杀死工作

$ $HADOOP_HOME/bin/hadoop job -kill <JOB-ID> 
e.g. 
$ $HADOOP_HOME/bin/hadoop job -kill job_201310191043_0004