Hadoop MapReduce编程开发

我就是我 2022-05-27 04:20 213阅读 0赞

用户编写完MapReduce程序后，按照一定的规则指定程序的输入和输出目录，并提交到Hadoop集群中。  
Hadoop将输入数据切分成若干个输入分片(input split)，并将每个split交给一个Map Task处理；Map Task不断的从对应的split中解析出一个个key/value，并调用map()函数处理。处理完之后根据Reduce Task个数将结果分成若干个分片(partition)写到本地磁盘；同时，每个Reduce Task从每个Map Task上读取属于自己的那个partition，然后基于排序的方法将key相同的数据聚集在一起，调用reduce()函数处理，并将结果输出到文件中。  
![这里写图片描述][70]

**Mapper类**

public class WordMap extends Mapper<Object, Text, Text, IntWritable> {
        @Override
        public void map(Object key, Text value, Context context) throw Exception {
            String line = value.toString();
            String[] arr = line.split("\t");
            for (String wd : arr) {
                // 每个单词出现１次，作为中间结果输出
                context.write(new Text(word), new IntWritable(1));
            }
        }
    }

**Reducer类**

public class WordReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throw Exception{
            int sum = 0;
            for (IntWritable count : values) {
                sum += count.get();
            }
            context.write(key, new IntWritable(sum));   // 输出最终结果
        }
    }

**main类**

public class WordCount {
        public static void main(String[] args) throws Exception {
            String intput = null;
            String output = null;
            if (null != args && args.length == 2) {
                Configuration conf = new Configuration();
                Job job = new Job(conf, “WordCount”);   // //创建一个job，并起名字
                job.setJarByClass(WordCount.class); // 以jar包的形式运行
    
                job.setMapperClass(WordMap.class);  // 设置处理Map/Reduce阶段的类
                job.setReducerClass(WordReduce.class);
    
                FileInputFormat.setInputPaths(job, new Path(args[0]));  //设置输入/输出路径
                FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
                job.setOutputKeyClass(Text.class);  //设置最终输出key/value的类型
                job.setOutputValueClass(IntWritable.class);
    
                System.exit(job.waitForCompletion(true) ? 0 : 1);   // 提交作业
            } else {
                System.err.println("<Urage> wordcount <intput> <output>");
            }
        }
    }

**集群运行**  
1. 打包作业  
单机上运行的程序不需要任何修改就可以直接在集群上运行，但是需要把程序打包为JAR文件发给集群。如果每个JAR文件都有一个作业，可以在JAR文件的manifest中指定要运行的main类。如果main类不在manifest中，则必须在命令行指定。

1.  启动作业  
    启动作业需要运行驱动程序，使用-conf选项来指定想要运行作业的集群（同样，也可以使用-fs和-jt选项）。如：hadoop jar uid.jar UidCollector.txt /sogou500w /out。job上的waitForCompletion()方法启动作业并检查进程，有任何变化，就输出一行map和reduce进度总结。
2.  通过web UI查看Job状态  
    Hadoop的Web界面用来浏览作业信息，对于跟踪作业运行进度、查找作业完成后的统计信息和日志非常有用。  
    关于集群的概要信息，包括集群的负载情况和使用情况：  
    当前正在运行的map和reduce数量，作业提交数量，可用的tasktracker节点数和集群的负载能力  
    集群中可用map和reduce的任务槽数(“Map Task Capacity”和“Reduce Task CapaciLy”)  
    每个节点平均可用的任务槽数等信息。
3.  获取结果  
    一旦作业完成，有许多方法可以获取结果（每个reducer产生一个输出文件）。因此在max-temp目录中会有30个部分文件（part file），命名为part-00000到part-00029。  
    如果输出文件很大，那么把文件分为多个part文件很重要，这样才能使多个reduce并行工作。如果文件采用这种分割形式，使用起来荣然很方便：例如作为另一个MapReduce作业的输入。在某些情况下，可以探索多个分割文件的结构来进行map端连接操作。

**远程调试**  
当一个任务失败且没有足够的记录信息来诊断错误时，可以用调试器运行该任务。在集群上运行作业时很难使用调试器，因为不知道哪个节点处理哪部分输入，所以不能在错误发生之前安装调试器。然而，有其他的方法可以进行调试。

1.  在本地重新产生错误  
    对于特定的输入，失败的任务通常总会失败。可以尝试通过下载致使任务失败的文件到本地运行进行问题重现，这样可以使用到调试器（如java的VisualVM）
2.  使用JVM调试选项  
    失败的常见原因是任务JVM中java内存溢出。可以将mapred.child.java.opts设置为 -xx:HeapDumpOnOutOfMemoryError-XX:HeapDumpPath=/path/to/dumps。该设置将会产生一个堆转储（heap dump）这可以通过Eclipse Memory Analyzer这样的工具来检查。注意该JVM选项应当添加到由mapred.child.java.opts指定的已有内存设置中。
3.  使用任务分析  
    java的profiler提供了很多JVM的内部细节，hadoop提供了分析作业中部分任务的机制。

**日志级别修改**  
打印日志的默认级别是INFO，因此DEBUG级别的消息不在syslog任务日志文件中出现。然而有时候又希望看到这些信息。集群中运行修改设置：mapreduce.map.log.level或者mapreduce.reduce.log.level  
例如：hadoop jar hadoop-examples.jar LoggingDriver -conf conf/hadoop-cluster.xml \\ -D mapreduce.map.log.level=DEBUG /sogou500w /logging-out

备注：有一些控制用于管理任务日志的大小和记录保留时间（默认日志最短在3小时候删除）  
通过yarn-nodemanager.log.retain-seconds属性来设置时间，如果日志聚合被激活，这个时间就可以忽略。  
为每个日志文件的最大规模设置一个阈值（默认值是0，表示没有上限）。通过mapreduce.task.userlog.limit.kb进行设置、

[70]: /images/20220527/873aa77420494a9fa70e4abdf48baa72.png