本文中的代码基于CDH的5.6版本编写
1、MrTest程序，模拟wordcount
输入文件：
hdfs文件路径：/tmp/test.txt 文件内容： hello world hello test test hadoop hadoop hdfs hive sql sqoop
代码：
package testhadoop;
import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
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;
//定义类
public class MRTest {
public static void main(String[] args){
Path inFile = new Path(args[0]);
Path outFile = new Path(args[1]);
Job job;
try {
job = Job.getInstance();
job.setJarByClass(testhadoop.MRTest.class);
//设置传入文件路径
FileInputFormat.addInputPath(job , inFile); //设置输出文件路径
FileOutputFormat.setOutputPath(job, outFile);
//设置Reducer使用的类
job.setReducerClass(testhadoop.MRreducer.class); //设置Mapper使用的类
job.setMapperClass(testhadoop.MRmapper.class);
//设置Mapper输出的key的class
job.setMapOutputKeyClass(Text.class); //设置Mapper输出的value的class
job.setMapOutputValueClass(IntWritable.class);
//设置Reducer输出的key的class
job.setOutputKeyClass(Text.class); //设置Reducer输出的value的class
job.setOutputValueClass(IntWritable.class);
try {
job.waitForCompletion(true);
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
class MRmapper extends Mapper<LongWritable, Text, Text, IntWritable>{
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub //将value转化成string类的数组
String[] line = value.toString().split(" ");
for(int i = 0 ; i< line.length-1 ;i++){
context.write(new Text(line[i]), new IntWritable(1));
}
}
}
class MRreducer extends Reducer<Text, IntWritable, Text, IntWritable>{
@Override
protected void reduce(Text arg0, Iterable<IntWritable> arg1,
Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
// TODO Auto-generated method stub
int sum = 0;
for(IntWritable i : arg1){
sum += i.get();
}
context.write(arg0, new IntWritable(sum));
}
}
导出成MRTest.jar并上传至服务器的/opt目录 执行： hadoop jar MRTest.jar "MRTest" "/tmp/test.txt" "/tmp/test/out" 再执行： hadoop fs -ls /tmp/test/out/* 查看输出的文件：
2、求最低温度
hdfs文件路径：/tmp/temp.txt 查看文件内容： 0067011990999991950051512004888888889999999N9+00221+9999999999999999999999 0067011990999991950051507004888888889999999N9+00001+9999999999999999999999 0067011990999991950051512004888888889999999N9+00221+9999999999999999999999 0067011990999991950051518004888888889999999N9-00111+9999999999999999999999 0067011990999991949032412004888888889999999N9+01111+9999999999999999999999 0067011990999991950032418004888888880500001N9+00001+9999999999999999999999 0067011990999991950051507004888888880500001N9+00781+9999999999999999999999
代码：
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.Mapper.Context;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class TempTest {
public static void main(String[] args) {
// TODO Auto-generated method stub
Path fromPath = new Path(args[0]);
Path toPath = new Path(args[1]);
try {
Configuration conf = new Configuration();
Job job = Job.getInstance();
job.setJarByClass(TempTest.class);
FileInputFormat.setInputPaths(job, fromPath);
FileOutputFormat.setOutputPath(job, toPath);
job.setMapperClass(TempMapper.class);
job.setReducerClass(TempReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
try {
job.waitForCompletion(true);
} catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
class TempMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
public static final int MISSING = 9999;
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
String line = value.toString();
String year = line.substring(15, 19);
String flag = line.substring(45,46);
int temp ;
if("+".equals(flag)){
temp = Integer.parseInt(line.substring(46,50));
}else{
temp = Integer.parseInt(line.substring(45,50));
}
String qua = line.substring(50,51);
if(temp != MISSING && qua.matches("[01459]")){
context.write(new Text(year), new IntWritable(temp));
}
}
}
class TempReducer extends Reducer<Text ,IntWritable ,Text,IntWritable>{
public static final Long MIN = Long.MIN_VALUE;
@Override
protected void reduce(Text arg0, Iterable<IntWritable> temp,
Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
// TODO Auto-generated method stub
int minValue = Integer.MAX_VALUE;
for(IntWritable t : temp){
minValue = Math.min(minValue, t.get());
}
context.write(arg0, new IntWritable(minValue));
}
}
导出成temp.jar并上传至服务器的/opt目录 执行： hadoop jar temp.jar "TempTest" "/tmp/temp.txt" "/tmp/temp/out" 再执行： hadoop fs -ls /tmp/temp/out/* 查看输出的文件：
时间： 2024-12-18 14:06:30