<!-- 自定义属性设置版本号 -->
<properties><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><hadoop-version>3.1.3</hadoop-version>
</properties>
<dependencies><!--junit单元测试--><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>RELEASE</version></dependency><!--slf4j日志--><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>2.0.0-alpha2</version></dependency><!--hadoop中的common工具类--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>${hadoop-version}</version></dependency><!--hadoop中的client--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop-version}</version></dependency><!--hadoop中的hdfs--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>${hadoop-version}</version></dependency>
</dependencies>

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

//Mapper类
//<>中的数据类型分别为输入key类型、输入value类型、输出key类型、输入value类型
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {private Text k=new Text();                   //输出keyprivate IntWritable v=new IntWritable(1);    //输出value，这里设置它恒为1//重载map方法，只要输入map后回车就可以自动生成方法框架@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString();     //获取一行数据String[] split = line.split(" ");   //切割，以空格为分隔符for (String word : split) {         k.set(word);            //设置输出的key为每一个单词context.write(k,v);     //写出，例如(北京, 1), (上海, 1)}}
}

//Reducer类
//<>中的数据类型分别为：
//输入key类型（Mapper输出key类型）、输入value类型（Mapper输出value类型）
//输出key类型（最终输出key类型）、输出value类型（最终输出value类型）
public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {private IntWritable v=new IntWritable();    //输出value//重载reduce方法@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int sum=0;    //每一个城市出现的次数for (IntWritable value : values) {  //求和，算出一个城市出现了几次sum+=value.get();}v.set(sum);   //输出value为每一个城市出现的次数context.write(key,v);    //写出}
}

//Driver类，用来调度Mapper和Reducer类，最终执行的就是这个类的main方法
public class WordCountDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//设置输入输出路径//要分析的txt放在input\wordcount文件夹下//输出路径wordcount是不存在的，执行main方法之后自动创建，如果该文件夹已经存在会报错args=new String[]{"D:\\input\\wordcount","D:\\output\\wordcount"};//写一个配置对象Configuration conf=new Configuration();//获取job对象Job job = Job.getInstance(conf);//设置jar位置job.setJarByClass(WordCountDriver.class);//关联mapper和reducerjob.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReducer.class);//设置mapper的输出类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//设置最终输出类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//设置输入输出路径FileInputFormat.setInputPaths(job,new Path(args[0]));FileOutputFormat.setOutputPath(job,new Path(args[1]));//提交boolean result = job.waitForCompletion(true);System.exit(result?0:1);}
}

<build><plugins><plugin><artifactId>maven-compiler-plugin</artifactId><version>3.6.1</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><artifactId>maven-assembly-plugin </artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs><archive><manifest><mainClass>com.mr.wordcount.WordCountDriver</mainClass></manifest></archive></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins>
</build>

sftp> cd /opt/module/hadoop-3.1.3

//com.mr.wordcount.WordCountDriver是Driver类的位置，记得更改
//将要统计分析的txt放在input文件夹下
//output文件夹必须不存在，执行后自动生成
[hadoop@hadoop101 hadoop-3.1.3]$ hadoop jar wordcount.jar com.mr.wordcount.WordCountDriver /input /output

//自定义序列化
public class StudentBean implements Writable {//对应txt中的几个变量和我们需要求的变量private String name;private long chinese;private long math;private long english;private long maxScore;private long minScore;private long sumScore;private double avgScore;//序列化@Overridepublic void write(DataOutput out) throws IOException {out.writeUTF(name);out.writeLong(chinese);out.writeLong(math);out.writeLong(english);out.writeLong(maxScore);out.writeLong(minScore);out.writeLong(sumScore);out.writeDouble(avgScore);}//反序列化@Overridepublic void readFields(DataInput in) throws IOException {name = in.readUTF();chinese = in.readLong();math = in.readLong();english = in.readLong();maxScore = in.readLong();minScore = in.readLong();sumScore = in.readLong();avgScore = in.readDouble();}//空参构造public StudentBean() {}//有参构造public StudentBean(String name, Long chinese, Long math, Long english) {this.name = name;this.chinese = chinese;this.math = math;this.english = english;this.maxScore = Math.max(Math.max(chinese, math), english);this.minScore = Math.min(Math.min(chinese, math), english);this.sumScore = (chinese + math + english);this.avgScore = this.sumScore / 3.0;}//以下这些右键-Generate-Getter & Setter就可以自动生成public String getName() {return name;}public void setName(String name) {this.name = name;}public Long getChinese() {return chinese;}public void setChinese(Long chinese) {this.chinese = chinese;}public Long getMath() {return math;}public void setMath(Long math) {this.math = math;}public Long getEnglish() {return english;}public void setEnglish(Long english) {this.english = english;}public Long getMaxScore() {return maxScore;}public void setMaxScore(Long maxScore) {this.maxScore = maxScore;}public Long getMinScore() {return minScore;}public void setMinScore(Long minScore) {this.minScore = minScore;}public Long getSumScore() {return sumScore;}public void setSumScore(Long sumScore) {this.sumScore = sumScore;}public Double getAvgScore() {return avgScore;}public void setAvgScore(Double avgScore) {this.avgScore = avgScore;}//重载toString方法，后续Mapper和Reducer会用到//不输出name是因为后续我们用name做key，输出的时候已经有name了，不需要重复输出@Overridepublic String toString() {return chinese +"\t" + math +"\t" + english +"\t" + maxScore +"\t" + minScore +"\t" + sumScore +"\t" + avgScore ;}
}

//Mapper类
public class StudentMapper extends Mapper<LongWritable, Text, Text, StudentBean> {private  Text k = new Text();private  StudentBean v = new StudentBean();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString(); //这是分割的另一种方法，可以按照\t,\n,空格等进行分割                    StringTokenizer sti = new StringTokenizer(line);    v.setName(sti.nextToken());v.setChinese(Long.parseLong(sti.nextToken()));v.setMath(Long.parseLong(sti.nextToken()));v.setEnglish(Long.parseLong(sti.nextToken()));k.set(v.getName());context.write(k, v);}
}

//Reducer类
public class StudentReducer extends Reducer<Text, StudentBean, Text, StudentBean> {@Overrideprotected void reduce(Text key, Iterable<StudentBean> values, Context context) throws IOException, InterruptedException {StudentBean bean = values.iterator().next();//计算在这步完成，我们编写的有参构造函数会帮我们完成计算StudentBean v = new StudentBean(bean.getName(), bean.getChinese(), bean.getMath(), bean.getEnglish());context.write(key, v);}
}

//在Mapper类中，获取上行流量与下行流量时
//因为不一定有ip地址，因此不能直接写split[3]
v.setUpFlow(Long.parseLong(split[split.length - 3]));
v.setDownFlow(Long.parseLong(split[split.length - 2]));

//Reducer类
public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {@Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {long sumUpFlow = 0;long sumDownFlow = 0;//对电话号码相同的用户，计算出其上下行流量的总和for (FlowBean bean : values) {          sumUpFlow += bean.getUpFlow();sumDownFlow += bean.getDownFlow();}FlowBean v = new FlowBean(key.toString(), sumUpFlow, sumDownFlow);context.write(key, v);}
}

//MySort类
public class MySort implements WritableComparable<MySort2> {private int age;private int height;//相等返回0，小于返回-1，大于返回1@Overridepublic int compareTo(MySort2 other) {if(this.age == other.age) {if(this.height == other.height) {return 0;}else if(this.height < other.height) {return  -1;}else {return 1;}}else if(this.age < other.age) {return -1;}else {return 1;}}//序列化@Overridepublic void write(DataOutput out) throws IOException {out.writeInt(age);out.writeInt(height);}//反序列化@Overridepublic void readFields(DataInput in) throws IOException {age = in.readInt();height = in.readInt();}public MySort() {}public MySort(int age, int height) {this.age = age;this.height = height;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}public int getHeight() {return height;}public void setHeight(int height) {this.height = height;}@Overridepublic String toString() {return age + height + "";}@Overridepublic int hashCode() {return age + height;}
}

//Mapper类中，将输出类型设置为MySort
private MySort k = new MySort();
k.setAge(v.getAge());
k.setHeight(v.getHeight());

//MyPartioner自定义分区
//泛型参数是mapper输出数据的key和value的数据类型
public class MyPartioner extends Partitioner<MySort, WorkSort> {@Overridepublic int getPartition(MySort key, WorkSort value, int num) {//按照年龄分区//num代表在年龄范围内平均分成几个区return value.getAge()/(100/num);}
}

//在Driver类中
//设置分区
job.setPartitionerClass(MyPartioner.class);
job.setNumReduceTasks(3);

//MyGroupingComparator类
public class MyGroupingComparator extends WritableComparator {@Overridepublic int compare(WritableComparable a, WritableComparable b) {//key1,key2是mapper输出的keyMySort2 key1 = (MySort2) a;MySort2 key2 = (MySort2) b;//身高在170以上的返回0，可以把key所对应的value组合放到同一key的values中if(key1.getHeight() > 170 && key2.getHeight() > 170)return  0;elsereturn -1;}public MyGroupingComparator() {//创建适合key的实例super(MySort2.class, true);}
}

//在Driver类中
//设置看哪些数据可以在reducer端进行一次处理
job.setGroupingComparatorClass(MyGroupingComparator.class);
job.setNumReduceTasks(1);

//Record类
public class Record implements Writable {private int isPerson;   //0表示score, 1表示personprivate String content; //输出内容//......
}

//MyJoin类
public class MyJoin {//Mapper类public static class JoinMapper extends Mapper<LongWritable, Text, IntWritable, Record> {private IntWritable k = new IntWritable();  //存放personIdprivate Record v = new Record();            //存放输出内容@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String [] fields = value.toString().split("\t");    //获取一行数据，以tab为分隔符int personId = 0;if(fields.length == 6) {    //如果长度为6，说明为person表personId = Integer.parseInt(fields[0]);     //person表的第一列为personIdv.setIsPerson(1);v.setContent(fields[1] + "\t" + fields[2] + "\t" + fields[3] + "\t" + fields[4] + "\t" + fields[5]);}else {  //否则是score表personId = Integer.parseInt(fields[1]);     //score表的第二列为personIdv.setIsPerson(0);v.setContent(fields[2] + "\t" + fields[3]);}k.set(personId);context.write(k, v);}}//Reducer类public static class JoinReducer extends Reducer<IntWritable, Record, IntWritable, Text> {private Text v = new Text();@Overrideprotected void reduce(IntWritable key, Iterable<Record> values, Context context) throws IOException, InterruptedException {String personStr = null;List<String> scoreList = new ArrayList();    //一个人可能有多门课的成绩for  (Record record: values) {if (record.getIsPerson() == 1){personStr = record.getContent();}else {scoreList.add(record.getContent());}}for (String score: scoreList) {v.set(personStr + "\t" + score);context.write(key, v);}}}//相当于Driver类public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//......Job job = Job.getInstance(conf, "queryMr");//......}
}

<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.19</version>
</dependency>

//Student类，注意一定要DBWritable类
public class Student implements DBWritable, Writable {//存放从mysql中读的数据private int sid;private String sname;private String sex;private int age;private String birthday;//序列化和反序列化的代码自行编写//从数据库中读取数据@Overridepublic void readFields(ResultSet resultSet) throws SQLException {this.sid = resultSet.getInt("sid");this.sname = resultSet.getString("sname");this.sex = resultSet.getString("sex");this.age = resultSet.getInt("age");this.birthday = resultSet.getString("birthday");}//其他方法自行编写
}

//ReadSql类，负责从MySql中读取数据
public class ReadSql {//Mapper类public static class ReadDBMapper extends Mapper<LongWritable, Student, IntWritable, IntWritable> {private IntWritable k = new IntWritable();private IntWritable v = new IntWritable(1);     //默认为1@Overrideprotected void map(LongWritable key, Student value, Context context) throws IOException, InterruptedException {k.set(value.getAge());    //按照需求，以年龄为keycontext.write(k, v);}}//Reducer类public static class ReadDBReducer extends Reducer<IntWritable, IntWritable, Text, NullWritable> {private Text k = new Text();@Overrideprotected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int count = 0;for (IntWritable value: values) {count += value.get();}k.set(key.get() + "岁的学生有" + count + "人");context.write(k, NullWritable.get());}}//相当于Driver类public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration conf = new Configuration();//连接数据库//localhost连接本地数据库，3306为端口，字符集为utf-8，时区为GMT（这一项不设置我会报错，大家可以自行调整）//最后两个参数是用户名和密码DBConfiguration.configureDB(conf, "com.mysql.jdbc.Driver", "jdbc:mysql://localhost:3306/mapreduce?characterEncoding=utf-8&serverTimezone=GMT", "root", "passwd");Job job = Job.getInstance(conf, "mrsql");//......//表中的属性名，不能出错，否则无法读取数据String[] fields = {"sid,", "sname", "sex", "age", "birthday"};//设置输入输出路径DBInputFormat.setInput(job, Student.class, "tb_student", null, "sid", fields);Path outPath = new Path("D:\\output\\sql");FileSystem fs = outPath.getFileSystem(conf);//如果输出路径存在，删除if(fs.exists(outPath)) {fs.delete(outPath, true);}FileOutputFormat.setOutputPath(job, outPath);//......}
}

//在Student类中
//写入数据库
@Override
public void write(PreparedStatement statement) throws SQLException {statement.setInt(1, sid);statement.setString(2, sname);statement.setString(3, sex);statement.setInt(4, age);statement.setString(5, birthday);
}

//WriteSql类，负责往MySql中写入数据
public class WriteSql {//Mapper类和Reducer类自行编写//相当于Driver类public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {//......//设置输入输出路径FileInputFormat.setInputPaths(job, new Path("D:\\input\\sql"));String[] fields = {"sid", "sname", "sex", "age", "birthday"};DBOutputFormat.setOutput(job, "tb_student", fields);//......}
}

public class Date {//Mapper类public static class DateFormatMapper extends Mapper<LongWritable, Text,Text, NullWritable> {private Text k=new Text();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//获取一行数据String text=new String(value.getBytes(),0,value.getLength(),"GBK");//以逗号分割，因为我们的文件是.csvString[] line = text.split(",", -1);//计算日期String date="2021年"+line[0];k.set(date+","+line[1]+","+line[2]+","+line[3]+","+line[4]+","+
line[5]+","+line[6]+","+line[7]);context.write(k,NullWritable.get());}}//Reducer和Driver类自行编写
}

public class Data implements WritableComparable<Data> {//定义6个变量用来存储6列数据//都设置为String是因为后四个数据都可能为空，不能设置为intprivate String date;private String city;private String district;private String sick;private String discharge;private String death;//重载比较方法，使最后数据能够按顺序输出@Overridepublic int compareTo(Data o) {int i = date.compareTo(o.date);int j = city.compareTo(o.city);int m = district.compareTo(o.district);int n = sick.compareTo(o.sick);int g = discharge.compareTo(o.discharge);int h = death.compareTo(o.death);if (i == 0) {if (j == 0) {if (m == 0) {if (n == 0) {if (g == 0) {return h;} else {return g;}} else {return n;}} else {return m;}} else {return j;}} else {return i;}}@Overridepublic boolean equals(Object o) {if (this == o)return true;if (o == null || getClass() != o.getClass())return false;Data data = (Data) o;return Objects.equals(date, data.date) &&Objects.equals(city, data.city) &&Objects.equals(district, data.district) &&Objects.equals(sick, data.sick) &&Objects.equals(discharge, data.discharge) &&Objects.equals(death, data.death);}@Overridepublic int hashCode() {return Objects.hash(date, city, district, sick, discharge, death);}//序列化和反序列化以及其他方法自行编写
}

public class Hubei {//过滤出湖北省的分区public static class DataPartioner extends Partitioner<Data, NullWritable> {//重写分区方法@Overridepublic int getPartition(Data data, NullWritable nullWritable, int numPartitions) {//比较开头是否为湖北，是湖北就符合条件，筛选出来if(data.getCity().startsWith("湖北")){return 0;}return 1;}}//Mapper, Reducer和Driver类自行编写//删除输出路径下的一切文件夹和文件public static void deleteFile(File file){File[] files = file.listFiles();if(files!=null&&files.length!=0){for (int i = 0; i < files.length; i++) {deleteFile(files[i]);}}file.delete();}public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {args = new String[]{"D:\\input\\covid19","D:\\output\\covid19.2"};File file=new File(args[1]);if(file.exists()){deleteFile(file);initDriver(args);}else{initDriver(args);}}
}