Hadoop笔记三: MapReduce

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
package cn.lazycat.bdd.hadoop.mapreduce.wordcount;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * Map作业类，需要继承Mapper(注意来自hadoop.mapreduce包)
 * 四个泛型分别表示输入key-value和输出key-value
 * 输入key是long，表示每行开始的偏移量
 * 输入value是Text，表示每行内容
 * 输出key是单词的内容
 * 输出value是单词在这一行出现的次数
 */
public class WordCountMapper
        extends Mapper<LongWritable, Text, Text, LongWritable> {

    /**
     * 用户定制的map方法
     * @param key 输入key
     * @param value 输入value
     * @param context 向外输出的对象
     */
    @Override
    protected void map(LongWritable key, Text value,
                       Context context)
            throws IOException, InterruptedException {

        // 获取读取的一行的文本的内容
        String line = value.toString();

        // 按照空格拆分，取得每一个单词
        String words[] = line.split(" ");

        for (String word : words) {
            // 对于每一个单词，产生一个输出
            context.write(new Text(word), new LongWritable(1));
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package cn.lazycat.bdd.hadoop.mapreduce.wordcount;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * Reduce作业类
 * 输入是Map作业结果的key-value类型，输入是最终的计算结果
 * 在本次案例，输入是每一个单词出现的标记，Reduce作业将
 * 这些次数做一个求和统计最终产生每个单词和单词出现的次数。
 */
public class WordCountReducer
        extends Reducer<Text, LongWritable, Text, LongWritable> {

    /**
     * reduce作业代码
     * @param key 来自map的key
     * @param values 洗牌后的values
     * @param context 输出结果
     */
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values,
                          Context context)
            throws IOException, InterruptedException {

        long sum = 0;   // 单词出现的次数
        for (LongWritable value : values) {
            sum += value.get();
        }

        // 写出数据
        context.write(key, new LongWritable(sum));
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
package cn.lazycat.bdd.hadoop.mapreduce.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;

public class WordCountJob {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // Hadoop 配置
        Configuration conf = new Configuration();
        // Hadoop Job 类
        // 传入conf对象和job的名称(均可缺省)
        Job job = Job.getInstance(conf, "wc");
        // 指定Job的入口类
        job.setJarByClass(WordCountJob.class);
        // 指定Mapper和Reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        // 指定Mapper的输出key-value类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        // 指定Reducer的输出key-value类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        // 指明输入的数据(来自HDFS)
        // 这里指定一个目录，Hadoop会读取目录下的所有文件(不会递归)
        FileInputFormat.addInputPath(job,
                new Path("hdfs://192.168.117.51:9000/park"));
        // 指明输出(输出到HDFS)
        // 注意这里不要事先创建，不然会抛出异常
        FileOutputFormat.setOutputPath(job,
                new Path("hdfs://192.168.117.51:9000/out"));
        // 开始执行任务
        job.waitForCompletion(true);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public class FlowMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    @Override
    protected void map(LongWritable key, Text value,
                       Context context)
            throws IOException, InterruptedException {

        String line = value.toString();

        // 取出三个需要的数据
        String data[] = line.split(" ");
        String phone = data[0];
        String name = data[2];
        int flow = Integer.parseInt(data[3]);

        context.write(new Text(phone + " " + name), new IntWritable(flow));

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class FlowReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {

        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }

        context.write(key, new IntWritable(sum));
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
package cn.lazycat.bdd.hadoop.mapreduce.flow.update;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class FlowBean implements Writable {

    private String phone;
    private String addr;
    private String name;
    private long flow;

    @Override
    public void write(DataOutput out) throws IOException {

        out.writeUTF(phone);
        out.writeUTF(addr);
        out.writeUTF(name);
        out.writeLong(flow);
    }

    @Override
    public void readFields(DataInput in) throws IOException {

        phone = in.readUTF();
        addr = in.readUTF();
        name = in.readUTF();
        flow = in.readLong();
    }

    // getter, setter 略
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public class UpdateFlowMapper
        extends Mapper<LongWritable, Text, Text, FlowBean> {

    @Override
    protected void map(LongWritable key, Text value,
                       Context context)
            throws IOException, InterruptedException {

        String line = value.toString();
        String temp[] = line.split(" ");

        String phone = temp[0];
        String addr = temp[1];
        String name = temp[2];
        long flow = Long.parseLong(temp[3]);

        FlowBean bean = new FlowBean();
        bean.setPhone(phone);
        bean.setAddr(addr);
        bean.setName(name);
        bean.setFlow(flow);

        context.write(new Text(phone), bean);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class UpdateFlowReduce
        extends Reducer<Text, FlowBean, Text, LongWritable> {
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values,
                          Context context)
            throws IOException, InterruptedException {

        String info = key.toString();
        long sum = 0;
        for (FlowBean flowBean : values) {
            sum += flowBean.getFlow();
        }

        context.write(new Text(info), new LongWritable(sum));
    }
}

1
reducer = (key.hashCode() & Integer.MAX_VALUE) % numberReduceTasks;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public class FlowPartitioner extends Partitioner<Text, FlowBean> {

    // 保存 城市-reducer编号
    private static Map<String, Integer> map;

    static {
        map = new HashMap<>();
        map.put("bj", 0);
        map.put("sh", 1);
        map.put("sz", 2);
    }

    @Override
    public int getPartition(Text text, FlowBean flowBean,
                            int numPartitions) {

        if (flowBean != null) {
            // 按照地址来分区
            String addr = flowBean.getAddr();
            return map.get(addr);
        }
        else {
            return 3;
        }

    }
}

1
2
3
4
// 设置自定义分区类
job.setPartitionerClass(FlowPartitioner.class);
// 设置分区数量
job.setNumReduceTasks(4);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// map:
@Override
protected void map(LongWritable key, Text value,
                    Context context)
        throws IOException, InterruptedException {

    String data[] = value.toString().split(" ");
    String name = data[1];
    long profit = Long.parseLong(data[2]) - Long.parseLong(data[3]);

    context.write(new Text(name), new LongWritable(profit));
}
// reduce:
@Override
protected void reduce(Text key, Iterable<LongWritable> values,
                        Context context)
        throws IOException, InterruptedException {

    long sum = 0;
    for (LongWritable val : values) {
        sum += val.get();
    }

    context.write(key, new LongWritable(sum));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
package cn.lazycat.bdd.hadoop.mapreduce.profit;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class ProfitBean
    implements WritableComparable<ProfitBean> {

    private String name;
    private long profit;

    @Override
    public int compareTo(ProfitBean o) {
        return (int) (o.getProfit() - profit);
    }

    @Override
    public void write(DataOutput out) throws IOException {

        out.writeUTF(name);
        out.writeLong(profit);

    }

    @Override
    public void readFields(DataInput in) throws IOException {

        name = in.readUTF();
        profit = in.readLong();
    }

    @Override
    public String toString() {
        return name + "\t" + profit;
    }

    // setter, getter 略

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// map
@Override
protected void map(LongWritable key, Text value,
                    Context context)
        throws IOException, InterruptedException {
    String data[] = value.toString().split(" ");
    ProfitBean bean = new ProfitBean();
    bean.setName(data[0]);
    bean.setProfit(Long.parseLong(data[1]));
    context.write(bean, NullWritable.get());
}
// reduce
@Override
protected void reduce(ProfitBean key, Iterable<NullWritable> values,
                        Context context)
        throws IOException, InterruptedException {

    context.write(new Text(key.toString()), NullWritable.get());

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
package cn.lazycat.bdd.hadoop.mapreduce.wordcount.combiner;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class WordCountCombiner
        extends Reducer<Text, LongWritable, Text, LongWritable> {


    @Override
    protected void reduce(Text key, Iterable<LongWritable> values,
                          Context context)
            throws IOException, InterruptedException {

        long sum = 0;
        for (LongWritable val : values) {
            sum += val.get();
        }

        context.write(key, new LongWritable(sum));
    }
}

1
2
// 指定Combiner
job.setCombinerClass(WordCountCombiner.class);

1
2
3
4
5
6
7
8
9
10
11
/**
 * 对文件进行逻辑切块。
 * 注意这没有进行物理切块，InputSplit包括了文件的位置、从哪开始切，到哪结束等信息。
 */
InputSplit[] getSplits(JobConf job);

/**
 * 指明了读取的方式，也就是一次读取多少内容，以及如何读取这些数据。
 * RecorderReader返回这次读取了多少内容，以及读取的Key和Value等。也就是读取的具体内容
 */
RecordReader<K, V> getRecordReader(InputSplit split, JobConf job);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
package cn.lazycat.bdd.hadoop.mapreduce.format;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;

import java.io.IOException;

public class ScoreRecordReader extends RecordReader<Text, Text> {

    // 读取文件的读行器
    private LineReader reader;

    // 每行的key
    private Text key;

    // 每行的value
    private Text value;

    // 因为一次性要读取多行，所以需要一个分隔符来区分不同行的数据
    private static String separator = "|";

    // 是否含有下一个数据
    private boolean hasNext = true;

    /**
     * 初始化RecordReader
     */
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {

        // 取得切块对象
        FileSplit fileSplit = (FileSplit) split;
        // 取得切块
        Path path = fileSplit.getPath();

        // 配置对象
        Configuration conf = new Configuration();
        // 取得文件系统对象
        FileSystem fs = path.getFileSystem(conf);
        // 在文件系统中取得要操作的文件流
        FSDataInputStream in = fs.open(path);
        // 创建读行器对象
        // BufferedReader reader = new BufferedReader(new InputStreamReader(in));
        reader = new LineReader(in);

    }

    /**
     * 判断是否存在下一个key-value。
     * 如果有，为key和value赋值。
     */
    @Override
    public boolean nextKeyValue() throws IOException {
        // 需要实例化一对key-value
        key = new Text();
        value = new Text();

        // 一次性读取4行
        Text cur = new Text();
        int len;
        for (int i = 0; i < 4; ++i) {
            len = reader.readLine(cur);
            if (len == 0) {  // 表示读取到最后一行了
                hasNext = false;   // 已经没有数据了
                break;
            }
            else {
                if (i == 0) {  // 4行中的第一行，应当是key值
                    key.set(cur);
                }
                else {        // 4行中的其余行，应该追加到value中
                    // 追加到尾端，以separator作为分割
                    value.append(cur.getBytes(), 0, cur.getLength());
                    value.append(separator.getBytes(), 0,
                            separator.length());
                }
            }
            cur.clear();
        }
        return hasNext;
    }

    /**
     * 返回当前的key
     */
    @Override
    public Text getCurrentKey() {
        return key;
    }

    /**
     * 返回当前的value
     */
    @Override
    public Text getCurrentValue() {
        return value;
    }

    /**
     * 返回当前的进度
     */
    @Override
    public float getProgress() {
        return hasNext ? 0f : 1f;
    }

    /**
     * 关闭执行的方法
     */
    @Override
    public void close() throws IOException {
        reader.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
package cn.lazycat.bdd.hadoop.mapreduce.format;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class ScoreInputFormat extends FileInputFormat<Text, Text> {
    @Override
    public RecordReader<Text, Text> createRecordReader(InputSplit split,
               TaskAttemptContext context) {
        return new ScoreRecordReader();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Override
protected void map(Text key, Text value, Context context)
        throws IOException, InterruptedException {

    String name = key.toString();
    double sum = 0;
    String data[] = value.toString().split("\\|");

    for (String str : data) {
        String scoreStr = str.split(" ")[1];
        sum += Double.parseDouble(scoreStr);
    }
    context.write(new Text(name), new DoubleWritable(sum));
}

1
job.setInputFormatClass(ScoreInputFormat.class);

1
2
3
4
// 指定数据来源以及InputFormat
addInputPath(job, path, inputFormatClass);
// 指定数据来源，InputFormat和Mapper
addInputPath(job, path, inputFormatClass, mapperClass);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
package cn.lazycat.bdd.hadoop.mapreduce.wordcount.format;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class WordCountRecordWriter extends RecordWriter<Text, LongWritable> {

    private FSDataOutputStream out;

    /**
     * 需要接收来自FS的输出流
     */
    public WordCountRecordWriter(FSDataOutputStream out) {
        this.out = out;
    }

    @Override
    public void write(Text key, LongWritable value) throws IOException {
        // 根据业务需求输出
        out.writeUTF(key.toString() + "-" + value + "#");
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException {
        out.close();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
package cn.lazycat.bdd.hadoop.mapreduce.wordcount.format;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCountOutputFormat extends FileOutputFormat<Text, LongWritable> {

    @Override
    public RecordWriter<Text, LongWritable> getRecordWriter(TaskAttemptContext job)
            throws IOException {
        // 获取conf对象
        Configuration conf = job.getConfiguration();
        // 获取path
        Path path = getDefaultWorkFile(job, "");
        // 获取文件系统对象
        FileSystem fs = path.getFileSystem(conf);

        FSDataOutputStream out = fs.create(path, false);

        return new WordCountRecordWriter(out);
    }
}

1
job.setOutputFormatClass(WordCountOutputFormat.class);

1
2
3
4
// 增加一个输出，需要指定标记和outputFormat
static addNamedOutput(job, "flag", outputFormat);
// 设定输出文件，flag表示输出文件的标记，key和value表示输出的内容
write("flag", key, value);

1
2
3
4
5
@Override
protected void setup(context) {
    super(context);
    outputs = new MultipleOutputs();
}

1
outputs.write("flag", key, value);

1
MultipleOutputs.addNamedOutput(job, "flag", outputFormat);

1
public int compare(byte[] b1, int s1, int l2, byte[] b2, int s2, int l2);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
package cn.lazycat.bdd.hadoop.mapreduce.wordcount.grouping;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparator;

import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.IOException;

public class WCGroupingComparator extends WritableComparator {

    @Override
    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
        Text key1 = new Text();
        Text key2 = new Text();

        // 对b1和b2进行反序列化
        deSerial(key1, b1, s1, l1);
        deSerial(key2, b2, s2, l2);

        // 如果key1和key2都是以a-n或者o-z开头的，认为它们相等
        if (chargeRegEquals(key1, key2, "[a-n][a-z]*")
                || chargeRegEquals(key1, key2, "[o-z][a-z]*")) {
            return 0;
        }
        else {  // 其它情况都认为不相等
            return -1;
        }
    }

    // 对字节数组进行反序列化
    private void deSerial(Text text, byte[] b, int s, int l) {
        DataInput in;
        try {
            in = new DataInputStream(new ByteArrayInputStream(b, s, l));
            // 反序列化
            text.readFields(in);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 判断两个text是否都符合某个正则表达式
    private boolean chargeRegEquals(Text text1, Text text2, String regex) {
        return text1.toString().matches(regex)
                && text2.toString().matches(regex);
    }
}

1
job.setGroupingComparatorClass(WCGroupingComparator.class);