MapReduce案例WordCount

環境準備

1.輸入檔案 a.txt

aa a
bbb cc
dd d dd ee
           

2.內建環境

參考之前部落格即可hadoop(阿裡雲内網環境)內建springboot及一些坑

map程式

package com.example.springbootintegrationtest.hadoop.mapreduce;

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;

/**
 * @author wuxinxin
 *
 * 對單詞做map操作,凡是map程式都要繼承Mapper
 */
public class WordCountMap extends Mapper<LongWritable,Text, Text, IntWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //資料是按照行擷取的 value,先按照分隔符進行拆分

        String s = value.toString();

        String[] split = s.split(" ");

        //拆分完後，開始做map操作
        Text k = new Text();
        IntWritable v = new IntWritable();
        v.set(1);
        for(String temp:split){
            //輸出map結果
            k.set(temp);
            context.write(k,v);
        }
    }
}

           

說明: map作用很簡單,就是将輸入的字元串,按照給定規則進行切割即可.例如我們統計每個單詞數量,那麼我們隻要根據以空格作為分隔符,對單詞一行行切割,輸出即可

reduce程式

package com.example.springbootintegrationtest.hadoop.mapreduce;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @author wuxinxin
 *
 * 對map結果做reduce
 * 凡是reduce程式,都需要繼承Reducer
 */
public class WordCountReduce extends Reducer<Text, IntWritable,Text,IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //對map好的結果進行合并操作（reduce）

        // key是代表的單詞， values代表數量
        int sum=0;

        for(IntWritable intWritable:values){
            sum+=intWritable.get();
        }

        //寫入reduce的結果
        context.write(key,new IntWritable(sum));
    }
}

           

說明: reduce程式就是對map切割好的資料進行一些統計和計算. 比如我們這裡是對每個單詞出現的次數進行計算,那麼寫我們的統計邏輯(對出現的相同單詞進行+1)即可.

driver(job)程式

package com.example.springbootintegrationtest.hadoop.mapreduce;

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;
import java.util.Arrays;
/**
 * 實作WordCount
 * @author wuxinxin
 */
public class WordCountDriver {

    public static void main(String[] args) throws Exception {

        /**
         * 建立一個job
         */
        Configuration conf=new Configuration();
        Job job = new Job(conf);

        //設定jar存儲位置
        job.setJarByClass(WordCountDriver.class);

        //關聯map和reduce程式
        job.setMapperClass(WordCountMap.class);
        job.setReducerClass(WordCountReduce.class);

        //設定map階段輸出key,value類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //設定最終輸出結果的key,value類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

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

        //送出job
        job.submit();
    }
}

           

說明:driver程式其實就是一個啟動程式,可以了解為map和reduce被包裝一個可執行程式,然後按照map和reduce邏輯進行計算. job其實是最小的執行單元,按照給定的輸入檔案,使用map進行分割資料,使用reduce做計算統計,最後輸出到給定的輸出結果路徑

執行結果

part-r-00000

a	1
aa	1
bbb	1
cc	1
d	1
dd	2
ee	1

           

說明: 目前程式使用的輸入輸出路徑都是本地路徑, 也可以直接使用hadoop的路徑,将輸入檔案和輸出結果都儲存在hadoop中.

總結

1.mapreduce是作為hadoop的計算元件,計算的資料可以來源于任何地方,一般而言還是來源于hdfs, 和hdfs配合使用

2. 我們這裡一定要清楚, mapreduce元件和hdfs元件是沒關系的, mapreduce計算資料可以來源任何地方.我們這裡舉例就來源于本地檔案資料.