实现过程所需的java代码

1，在hdfs目录/tmp/input/wordcount中有一系列文件，内容均为","号分隔，

求按","号分隔的各个元素的出现频率，输出到目录/tmp/个人用户名的hdfs目录中。

实现代码：

package com.tledu.hadoop.mr.homework;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.StringTokenizer;public class HomeWork1 {public static class HomeWork1Mapper extends Mapper<Object, Text, Text, IntWritable> {Text word = new Text();IntWritable one = new IntWritable(1);@Overrideprotected void map(Object key, Text value, Context context) throws IOException, InterruptedException {StringTokenizer str = new StringTokenizer(value.toString(), ",");while (str.hasMoreTokens()) {word.set(str.nextToken());context.write(word, one);}}}public static class HomeWork1Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {IntWritable sumRes = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int sum = 0;for (IntWritable val : values) {sum += val.get();}sumRes.set(sum);context.write(key, sumRes);}}public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {// 1. 创建配置Configuration conf = new Configuration();// 2. 创建任务Job job = Job.getInstance(conf, "kxr-homework1");// 3. 设置对应类job.setJarByClass(HomeWork1.class);job.setMapperClass(HomeWork1Mapper.class);job.setCombinerClass(HomeWork1Reducer.class);job.setReducerClass(HomeWork1Reducer.class);// 4. 输出的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);// 5. 输入和输出的路径// 获取所有输入的txt文件List<String> fileList = getTxtFileListFromPath(args[0]);for (String filePath : fileList) {FileInputFormat.addInputPath(job, new Path(filePath));}FileOutputFormat.setOutputPath(job, new Path(args[1]));// 6. 启动任务System.exit(job.waitForCompletion(true)?0:1);}/*** 从指定路径中递归读取所有的txt文件** @param folderPath 文件夹的地址* @return 返回txt文件地址列表*/public static List<String> getTxtFileListFromPath(String folderPath) throws IOException {Configuration conf = new Configuration();FileSystem fs = FileSystem.get(conf);Path path = new Path(folderPath);// 1. 获取文件夹下所有的文件FileStatus[] statuses = fs.listStatus(path);// 对应的txt文件的地址列表List<String> list = new ArrayList<>();for (FileStatus fileStatus : statuses) {// 获取到对应的文件Path file = fileStatus.getPath();// 读取的是txt文件if (fileStatus.isFile() && file.getName().endsWith(".txt")) {list.add(file.toString());}// 如果是文件夹else if (fileStatus.isDirectory()) {list.addAll(getTxtFileListFromPath(file.toString()));}}return list;}
}

2、在hdfs目录/tmp/input/wordcount目录中有一系列文件，内容为","号分隔，分隔后的元素均为数值类型、字母、中文，求所有出现的数值的平均值

实现代码：

package com.tledu.hadoop.mr.homework;import com.tledu.hadoop.utils.RegUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.StringTokenizer;public class HomeWork3 {public static class HomeWork3Mapper extends Mapper<Object, Text, Text, DoubleWritable> {Text word = new Text("sum=");DoubleWritable number = new DoubleWritable();@Overrideprotected void map(Object key, Text value, Context context) throws IOException, InterruptedException {String path = ((FileSplit)context.getInputSplit()).getPath().toString();//最后一个斜线的位置int index = path.lastIndexOf("/");//得到输入文件的文件名String fileName = path.substring(index+1);StringTokenizer str = new StringTokenizer(value.toString(), ",");while (str.hasMoreTokens()) {String text = str.nextToken();if (RegUtils.isNumber(text)) {number.set(Double.parseDouble(text));context.write(word, number);}}}}public static class HomeWork3Reducer extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {DoubleWritable sumRes = new DoubleWritable();@Overrideprotected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {double sum = 0;int count = 0;for (DoubleWritable val : values) {sum += val.get();count++;}System.out.println(sum / count);sumRes.set(sum / count);context.write(key, sumRes);}}public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {// 1. 创建配置Configuration conf = new Configuration();// 2. 创建任务Job job = Job.getInstance(conf, "kxr-homework2");// 3. 设置对应类job.setJarByClass(HomeWork3.class);job.setMapperClass(HomeWork3Mapper.class);
//        job.setCombinerClass(HomeWork3Reducer.class);job.setReducerClass(HomeWork3Reducer.class);// 4. 输出的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(DoubleWritable.class);// 5. 输入和输出的路径// 获取所有输入的txt文件List<String> fileList = getTxtFileListFromPath(args[0]);for (String filePath : fileList) {FileInputFormat.addInputPath(job, new Path(filePath));}FileOutputFormat.setOutputPath(job, new Path(args[1]));// 6. 启动任务System.exit(job.waitForCompletion(true)?0:1);}/*** 从指定路径中递归读取所有的txt文件** @param folderPath 文件夹的地址* @return 返回txt文件地址列表*/public static List<String> getTxtFileListFromPath(String folderPath) throws IOException {Configuration conf = new Configuration();FileSystem fs = FileSystem.get(conf);Path path = new Path(folderPath);// 1. 获取文件夹下所有的文件FileStatus[] statuses = fs.listStatus(path);// 对应的txt文件的地址列表List<String> list = new ArrayList<>();for (FileStatus fileStatus : statuses) {// 获取到对应的文件Path file = fileStatus.getPath();// 读取的是txt文件if (fileStatus.isFile() && file.getName().endsWith(".txt")) {list.add(file.toString());}// 如果是文件夹else if (fileStatus.isDirectory()) {list.addAll(getTxtFileListFromPath(file.toString()));}}return list;}
}

3，在hdfs目录/tmp/input/wordcount目录中有一系列文件，内容为","号分隔，同时在hdfs路径/tmp/black.txt黑名单文件，一行一个单词用于存放不记入统计的单词列表。求按","号分隔的各个元素去除掉黑名单后的出现频率，输出到目录/tmp/output/个人用户名的hdfs目录中。

实现代码：

package com.tledu.hadoop.mr.homework;import com.tledu.hadoop.utils.RegUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.StringTokenizer;public class HomeWork5 {public static class HomeWork5Mapper extends Mapper<Object, Text, Text, IntWritable> {Text word = new Text();IntWritable one = new IntWritable(1);IntWritable blackVal = new IntWritable(0);@Overrideprotected void map(Object key, Text value, Context context) throws IOException, InterruptedException {String path = ((FileSplit) context.getInputSplit()).getPath().toString();int index = path.lastIndexOf("/");//得到输入文件的文件名String fileName = path.substring(index+1);if (fileName.contains("blacklist")) {// 是黑名单的内容word.set(value.toString());context.write(word,blackVal);}else {StringTokenizer str = new StringTokenizer(value.toString(), ",");while (str.hasMoreTokens()) {word.set(str.nextToken());context.write(word, one);}}}}public static class HomeWork5Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {IntWritable sumRes = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int sum = 0;boolean isBlack = false;for (IntWritable val : values) {// 有一项等于0了，说明它属于黑名单if (val.get() == 0) {// 这是黑名单isBlack = true;break;}sum += val.get();}sumRes.set(sum);if (!isBlack) {context.write(key, sumRes);}}}public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {// 1. 创建配置Configuration conf = new Configuration();GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);String[] remainingArgs = optionParser.getRemainingArgs();// 2. 创建任务Job job = Job.getInstance(conf, "kxr-homework5");// 3. 设置对应类job.setJarByClass(HomeWork5.class);job.setMapperClass(HomeWork5Mapper.class);
//        job.setCombinerClass(HomeWork5Reducer.class);job.setReducerClass(HomeWork5Reducer.class);// 4. 输出的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);// 5. 输入和输出的路径// 获取所有输入的txt文件List<String> fileList = getTxtFileListFromPath(remainingArgs[0]);for (String filePath : fileList) {FileInputFormat.addInputPath(job, new Path(filePath));}FileOutputFormat.setOutputPath(job, new Path(remainingArgs[1]));// 6. 启动任务System.exit(job.waitForCompletion(true)?0:1);}/*** 从指定路径中递归读取所有的txt文件** @param folderPath 文件夹的地址* @return 返回txt文件地址列表*/public static List<String> getTxtFileListFromPath(String folderPath) throws IOException {Configuration conf = new Configuration();FileSystem fs = FileSystem.get(conf);Path path = new Path(folderPath);// 1. 获取文件夹下所有的文件FileStatus[] statuses = fs.listStatus(path);// 对应的txt文件的地址列表List<String> list = new ArrayList<>();for (FileStatus fileStatus : statuses) {// 获取到对应的文件Path file = fileStatus.getPath();// 读取的是txt文件if (fileStatus.isFile() && file.getName().endsWith(".txt")) {list.add(file.toString());}// 如果是文件夹else if (fileStatus.isDirectory()) {list.addAll(getTxtFileListFromPath(file.toString()));}}return list;}
}