您现在的位置是:首页 >技术教程 >Hadoop-----WorldCount代码编写、温度案例网站首页技术教程
Hadoop-----WorldCount代码编写、温度案例
简介Hadoop-----WorldCount代码编写、温度案例
WorldCount代码编写
WordCountMapper
package day34.com.doit.demo02;
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;
public class WordCountMapper extends Mapper<LongWritable , Text ,Text, IntWritable> {
/**
*
* @param key
* @param value
* @param context
* @throws IOException
* @throws InterruptedException
*/
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
String[] wordArr = value.toString().split("\s+");
for (String word : wordArr) {
context.write(new Text(word),new IntWritable(1));
}
}
}
WordCountReducer
package day34.com.doit.demo02;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {
@Override
protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable i : values) {
sum+=i.get();
}
context.write(key,new IntWritable(sum));
}
}
Test
package day34.com.doit.demo02;
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;
public class Test {
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
Configuration con = new Configuration();
Job job = Job.getInstance(con,"wordCount");
//设置Mapper 和Reducer
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);
//Mapper的输出
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//Reducer的输出
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setNumReduceTasks(2);
//设置输入 输出路径
FileInputFormat.setInputPaths(job,new Path("d:\work\abc\input"));
FileOutputFormat.setOutputPath(job,new Path("d:\work\abc\output5"));
job.waitForCompletion(true);
}
}
温度案例
同一时间不同地区的温度 求每天的最高温度
2022-04-03,21.2
2022-04-03,18.5
2022-04-03,24.3
2022-04-03,16.5
2022-04-03,10.0
2022-04-04,28.3
2022-04-04,18.7
2022-04-04,30.0
2022-04-04,21.1
代码实现
package day34.com.doit.demo03;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
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;
public class Test {
private static class TemperatureMapper extends Mapper<LongWritable, Text,Text, DoubleWritable> {
private Text k2 = new Text();
private DoubleWritable v2 = new DoubleWritable();
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
String line = value.toString();
String[] arr = line.split(",");
k2.set(arr[0]);
v2.set(Double.parseDouble(arr[1]));
context.write(k2,v2);
}
}
private static class TemperatureReducer extends Reducer<Text,DoubleWritable,Text,DoubleWritable> {
private DoubleWritable d = new DoubleWritable();
@Override
protected void reduce(Text key, Iterable<DoubleWritable> values, Reducer<Text, DoubleWritable, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
double max = values.iterator().next().get();
for (DoubleWritable value : values) {
double v = value.get();
if (v>max) {
max = v;
}
}
d.set(max);
context.write(key,d);
}
}
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
Configuration con = new Configuration();
Job job = Job.getInstance(con,"maxTemperature");
//设置Mapper和Reducer
job.setMapperClass(TemperatureMapper.class);
job.setReducerClass(TemperatureReducer.class);
//设置Map输出类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(DoubleWritable.class);
//设置reduce的输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(DoubleWritable.class);
//设置输入文件位置
FileInputFormat.setInputPaths(job,new Path("d:\work\abc\temprature.txt"));
FileOutputFormat.setOutputPath(job,new Path("d:\work\abc\out_put"));
//将任务提交 并等待完成
job.waitForCompletion(true);
}
}
风语者!平时喜欢研究各种技术,目前在从事后端开发工作,热爱生活、热爱工作。