MapReduce工作机制——Word Count实例(一)

MapReduce工作体制——Word Count实例(一)

MapReduce的惦记是分布式计算,也便是分而治之,并行总括提升速度。

编制程序思想

第2,要将数据抽象为键值对的款式,map函数输入键值对,处理后,发生新的键值对作为中间结果输出。接着,MapReduce框架自动将中间结果按键做聚合处理,发给reduce函数处理。末了,reduce函数以键和对应的值的集聚作为输入,处理后,发生另一多种键值对作为最后输出。后边会构成实例介绍任何经过。

运作条件

先不考虑使用YA帕杰罗N的情景,那多少个时候MapReduce的运营环境便是YAHavalN,此处大家谈谈的是上一世环境。

TaskTracker

slave的剧中人物,负责申报心跳和执行命令。1个集群有四个TaskTracker,但2个节点只有1个,TaskTracker和DataNode运维在同一节点。

JobTracker

master的角色,负义务务调度和集群财富监察,不参预计算。依据TaskTracker周期性发来的心跳信息,考虑TaskTracker的能源剩余量、作业优先级等等,为其分合营适的任务。

Word Count实例

环境

  • Java 1.7
  • Hadoop 2.7
  • Maven 3.3
  • Intellij IDEA 2016.3
  • Windows 10

题主在合龙开发条件下写了Word Count程序,配置的pom.xml如下:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>WordCount</groupId>
    <artifactId>Hadoop</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>2.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-common</artifactId>
            <version>2.7.0</version>
        </dependency>

    </dependencies>

</project>

编码

  1. Mapper类

    Mapper类的陆个泛型分别代表:map函数输入键值对的键的类,map函数输入键值对的值的类,map函数输出键值对的键的类,map函数输出键值对的值的类。

    map函数部分,key是LongWritable类型,表示该行;value是Text类型,表示行的始末;Context类的write(Text
    key, IntWritable value)将中等结果输出。

    package com.hellohadoop;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    import java.util.StringTokenizer;
    
    /**
     * Created by duyue on 2017/7/13.
     */
    public class TokenizerMapper  extends Mapper<LongWritable, Text, Text, IntWritable> {
    
        // 直接把单词的个数设置成1, 认为出现了1次
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
    
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 每行文本拆分成单个单词
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                // 每个单词(忽略重复)的个数都为1
                // 即,出现两次"good"会写入两次"good",而不会认为"good"出现了2次
                context.write(word, one);
            }
        }
    }
    
  2. Reducer类

    Reducer类的5个泛型表示:reduce函数输入键值对的键的类,reduce函数输入键值对的值的类(与map函数输出对应),reduce函数输出键值对的键的类,reduce函数输出键值对的值的类。

    reduce函数部分:接收到的参数形如:<key, List<value>>,因为map函数把key值相同(同一单词)的具备value都发送给reduce函数,总括后输出结果。

    package com.hellohadoop;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    import java.util.Iterator;
    
    /**
     * Created by duyue on 2017/7/13.
     */
    public class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    
        private IntWritable result = new IntWritable();
    
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }
    
  3. 编写main函数

    package com.hellohadoop;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    /**
     * Created by duyue on 2017/7/13.
     */
    public class WordCount {
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            Configuration conf = new Configuration();
            Job job = new Job(conf, "word count");
            job.setJarByClass(WordCount.class);
            job.setMapperClass(TokenizerMapper.class);
            job.setReducerClass(IntSumReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            FileInputFormat.addInputPath(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
    }
    

运作程序

此处首要借助于事先Maven重视的包,为了成功呈现日志文件,须要在resources包中添加log4j.properties文本,地点如下图:

log4j.properties配置.PNG

文本配置:

log4j.rootLogger=debug, stdout, R

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

log4j.appender.R=org.apache.log4j.RollingFileAppender
log4j.appender.R.File=firestorm.log

log4j.appender.R.MaxFileSize=100KB
log4j.appender.R.MaxBackupIndex=1

log4j.appender.R.layout=org.apache.log4j.PatternLayout
log4j.appender.R.layout.ConversionPattern=%p %t %c - %m%n

log4j.logger.com.codefutures=DEBUG

配置Configuration如图:

Configuration配置.png

团结要创制input文件夹,并将在Project
Structure中安装为Excluded类型。在input文件夹下成立必要计算单词数的文书,地点如下图:

Input文件夹地方

题主总计的是Shakespeare有名的十四行诗的Sonnet 18,运营程序后:

运作后的状态图

其中part-r-00000中保留了计算结果,图太长,截了一部分:

总结结果

以上正是Word Count实例在Idea下运维的景象

下期预先报告:MapReduce编制程序涉及到的API

相关文章