1. 程式人生 > >使用 IntelliJ IDEA寫出你的第一個mapruduce應用

使用 IntelliJ IDEA寫出你的第一個mapruduce應用

    文章基於widows環境下的idea,hadoop版本2.7.2,案例來源於《Hadoop:The Definitive Guide》第二章,需要統計美國自1901年以來每一年的最高氣溫,資料檔案見百度盤 https://pan.baidu.com/s/1gfoZkUf

1、配置maven

2、解讀(簡版,詳見book)

a.資料解讀

資料格式

    從資料格式中可以看到每一行(15,19)是年份資訊,而(87,92)儲存的是溫度,(92,93)儲存的是天氣的質量資訊,我們需要提取出年份和溫度資訊。

b.map階段

    在map階段只是一個準備資料的過程,提取年份和溫度形成鍵值對輸出給reduce階段;並且在這個過程中篩掉缺失和錯誤的資料。

map結果

    map結果提取出每一行對應的年份和溫度資訊。

c.reduce階段

    map函式的輸出經過MapReduce框架處理後,最後傳送到reduce函式,這個過程基於鍵來對鍵值對進行進行排序和分組,因此,reduce函式看到的是如下輸入:

這裡寫圖片描述
每一年緊接著一系列氣溫資料,reduce要做的工作就是遍歷整個列表並找出最大的數值。
這裡寫圖片描述

3、code

a.map階段—獲取每一年的溫度,以(key,value)鍵值對儲存

package mkd.hadoop.mr;
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; /** * Created by MKD on 2017/7/12. * map階段 */ public class MyMaxTempMapper extends Mapper<LongWritable,Text,Text,IntWritable>{ private static final int MISSING = 9999
; @Override public void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{ String line = value.toString();//將每一行資料轉化成String String year = line.substring(15,19);//獲取年份資料 int airTemperature; //提取氣溫 + / - if(line.charAt(87) == '+'){ airTemperature = Integer.parseInt(line.substring(88,92));//將string轉化成int }else{ airTemperature = Integer.parseInt(line.substring(87,92)); } //質量 String quality = line.substring(92,93); //判斷氣溫有效性 if(airTemperature != MISSING && quality.matches("[01459]")){ context.write(new Text(year),new IntWritable(airTemperature)); } } }

b.reduce階段—找出每一年最大溫度

package mkd.hadoop.mr;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;

/**
 * Created by MKD on 2017/7/12.
 * reduce階段
 */
public class MyMaxTemperReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
    @Override
    public void reduce(Text key,Iterable<IntWritable> values,Context context)
        throws IOException,InterruptedException{
        int maxValue = Integer.MIN_VALUE;
        for(IntWritable value:values){//獲取最大溫度
            maxValue = Math.max(maxValue,value.get());
        }
        context.write(key,new IntWritable(maxValue));
    }
}

c.MapReduceJob

package mkd.hadoop.mr;
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;

/**
 * Created by MKD on 2017/7/12.
 * mapreduce APP
 */
public class MyMaxTemputer {
    public static void main(String[] args) throws Exception{
        if(args.length != 2){
            System.err.println("Usage:MaxTemperture <input path> <output path>");
            System.exit(-1);
        }
        Job job = Job.getInstance();
        job.setJarByClass(MyMaxTemputer.class);
        job.setJobName("MyMaxTemputer");//設定作業名稱

        FileInputFormat.addInputPath(job,new Path(args[0]));//輸入路徑
        FileOutputFormat.setOutputPath(job,new Path(args[1]));//輸出路徑

        job.setMapperClass(MyMaxTempMapper.class);//設定mapper型別
        job.setReducerClass(MyMaxTemperReducer.class);//設定reduce型別

        job.setOutputKeyClass(Text.class);//設定輸出key型別
        job.setOutputValueClass(IntWritable.class);//設定輸出value型別

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

4、設定輸入輸出

這裡寫圖片描述

    上面是輸入路徑,下面是輸出路徑(輸出目錄一定要不存在!!!!,否則報錯)

5、執行APP

這裡寫圖片描述

    然後在輸出路徑下可以看到如下檔案,_SUCCESS表示執行成功,我們開啟part-r-0000檔案可以檢視結果

這裡寫圖片描述