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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
package club.kwcoder.covid;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Objects;
public class CovidWritable implements WritableComparable<CovidWritable> {
private String state;
private String county;
private Integer cases;
private Integer deaths;
@Override
public int compareTo(CovidWritable o) {
// 若比较对象为null，则自身大
if (null == o) {
return 1;
}
// 若州不一样，则比较州
if (!this.state.equals(o.state)) {
return this.state.compareTo(o.state);
}
// 若州一样，则比较县
return this.county.compareTo(o.county);
// map()结束时总确诊数还没有计算，因此没必要比较
}
/**
* Hadoop序列化
* @param out <code>DataOuput</code> to serialize this object into.
* @throws IOException
*/
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(state);
out.writeUTF(county);
out.writeInt(cases);
out.writeInt(deaths);
}
/**
* Hadoop反序列化
* @param in <code>DataInput</code> to deseriablize this object from.
* @throws IOException
*/
@Override
public void readFields(DataInput in) throws IOException {
this.state = in.readUTF();
this.county = in.readUTF();
this.cases = in.readInt();
this.deaths = in.readInt();
}
/**
* 建造者模式构建对象
*/
public static class Builder {
private String state;
private String county;
private Integer cases;
private Integer deaths;
public Builder setState(String state) {
this.state = state;
return this;
}
public Builder setCounty(String county) {
this.county = county;
return this;
}
public Builder setCases(Integer cases) {
this.cases = cases;
return this;
}
public Builder setDeaths(Integer deaths) {
this.deaths = deaths;
return this;
}
Builder() {
}
CovidWritable build() {
return new CovidWritable(this);
}
}
public CovidWritable(Builder builder) {
this.state = builder.state;
this.county = builder.county;
this.cases = builder.cases;
this.deaths = builder.deaths;
}
@Override
public String toString() {
return state + 't' + county + 't' + "t" + cases + "t" + deaths;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
CovidWritable that = (CovidWritable) o;
if (!Objects.equals(state, that.state)) return false;
if (!Objects.equals(county, that.county)) return false;
if (!Objects.equals(cases, that.cases)) return false;
return Objects.equals(deaths, that.deaths);
}
@Override
public int hashCode() {
int result = state != null ? state.hashCode() : 0;
result = 31 * result + (county != null ? county.hashCode() : 0);
result = 31 * result + (cases != null ? cases.hashCode() : 0);
result = 31 * result + (deaths != null ? deaths.hashCode() : 0);
return result;
}
public String getState() {
return state;
}
public void setState(String state) {
this.state = state;
}
public String getCounty() {
return county;
}
public void setCounty(String county) {
this.county = county;
}
public Integer getCases() {
return cases;
}
public void setCases(Integer cases) {
this.cases = cases;
}
public Integer getDeaths() {
return deaths;
}
public void setDeaths(Integer deaths) {
this.deaths = deaths;
}
public CovidWritable(String state, String county, Integer cases, Integer deaths) {
this.state = state;
this.county = county;
this.cases = cases;
this.deaths = deaths;
}
public CovidWritable() {
}
}

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
package club.kwcoder.covid;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class CovidMapper extends Mapper<LongWritable, Text, CovidWritable, NullWritable> {
private final CovidWritable.Builder builder = new CovidWritable.Builder();
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, CovidWritable, NullWritable>.Context context) throws IOException, InterruptedException {
String line = value.toString();
// 排除空行
if (StringUtils.isBlank(line)) {
return;
}
// 排除首行
if (key.get() == 0L && line.startsWith("date")) {
return;
}
// 切分与排除非法数据
String[] items = line.split(",", 6);
if (items.length != 6) {
return;
}
for (String item : items) {
if (StringUtils.isBlank(item)) {
return;
}
}
// 构建对象并输出
CovidWritable outKey = builder
.setState(items[2])
.setCounty(items[1])
.setCases(Integer.parseInt(items[4]))
.setDeaths(Integer.parseInt(items[5])).build();
context.write(outKey, NullWritable.get());
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
package club.kwcoder.covid;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class CovidCombinerGrouping extends WritableComparator {
public CovidCombinerGrouping() {
super(CovidWritable.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
// 当属于同一县时，归位一组
return ((CovidWritable) a).getCounty().compareTo(((CovidWritable) b).getCounty());
}
}

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
package club.kwcoder.covid;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class CovidCombiner extends Reducer<CovidWritable, NullWritable, CovidWritable, NullWritable> {
private final CovidWritable.Builder outKeyBuild = new CovidWritable.Builder();
@Override
protected void reduce(CovidWritable key, Iterable<NullWritable> values, Reducer<CovidWritable, NullWritable, CovidWritable, NullWritable>.Context context) throws IOException, InterruptedException {
String state = "", county = "";
Integer cases = 0, deaths = 0;
// 累加确诊数、死亡数
for (NullWritable ignored : values) {
if (state.equals("")) {
state = key.getState();
county = key.getCounty();
}
cases += key.getCases();
deaths += key.getDeaths();
}
// 构建对象并输出
CovidWritable build = outKeyBuild
.setState(state)
.setCounty(county)
.setCases(cases)
.setDeaths(deaths)
.build();
context.write(build, NullWritable.get());
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
package club.kwcoder.covid;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class CovidSortByCaseASC extends WritableComparator {
public CovidSortByCaseASC() {
super(CovidWritable.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
CovidWritable a1 = (CovidWritable) a;
CovidWritable b1 = (CovidWritable) b;
// 当州不同时，根据州名的字典顺序排序
if (!a1.getState().equals(b1.getState())) {
return a1.getState().compareTo(b1.getState());
}
// 当州相同时，根据确诊数降序排序
return -Integer.compare(((CovidWritable) a).getCases(), ((CovidWritable) b).getCases());
}
}

1
2
3
4
5
6
7
8
9
10
package club.kwcoder.covid;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class CovidReducer extends Reducer<CovidWritable, NullWritable, CovidWritable, NullWritable> {
@Override
protected void reduce(CovidWritable key, Iterable<NullWritable> values, Reducer<CovidWritable, NullWritable, CovidWritable, NullWritable>.Context context) throws IOException, InterruptedException {
context.write(key, 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
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
package club.kwcoder.covid;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;
import java.io.IOException;
public class CovidRunner {
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
Configuration conf = new Configuration();
FileSystem hdfs = FileSystem.get(conf);
Path input = new Path("/covid");
Path output = new Path("/covid_result");
if (hdfs.exists(output)) {
hdfs.delete(output, true);
}
// 配置Job
Job job = Job.getInstance(conf, "covid");
job.setJarByClass(CovidRunner.class);
// 配置输入
job.setInputFormatClass(TextInputFormat.class);
FileInputFormat.setInputPaths(job, input);
// 配置Mapper类
job.setMapperClass(CovidMapper.class);
job.setMapOutputKeyClass(CovidWritable.class);
job.setMapOutputValueClass(NullWritable.class);
// 设置Combiner阶段的分组
job.setCombinerKeyGroupingComparatorClass(CovidCombinerGrouping.class);
// 设置Combiner
job.setCombinerClass(CovidCombiner.class);
// 设置自定义排序
job.setSortComparatorClass(CovidSortByCaseASC.class);
// 配置Reducer类
job.setReducerClass(CovidReducer.class);
job.setOutputKeyClass(CovidWritable.class);
job.setOutputValueClass(NullWritable.class);
// 配置输出
job.setOutputFormatClass(TextOutputFormat.class);
FileOutputFormat.setOutputPath(job, output);
// 运行
boolean flag = job.waitForCompletion(true);
if (flag) {
System.out.println("covid process success");
}
}
}