大數據實驗（四）MapReduce編程實踐（Ubuntu)

大數據實驗（四）MapReduce編程實踐（Ubuntu)

前置工具及環境

Ubuntu 16.4

VirtualBox

Hadoop 2.7.3

jdk1.8

一、MapReduce簡介

MapReduce是Hadoop提供的一個分布式計算框架，MapReduce 作業通過將輸入的數據集拆分為獨立的塊，這些塊由 map 以並行的方式處理，框架對 map 的輸出進行排序，然后輸入到 reduce 中。MapReduce 框架專門用於 <key，value> 鍵值對處理，它將作業的輸入視為一組 <key，value> 對，並生成一組 <key，value> 對作為輸出。

過程大概如此：

(input) <k1, v1> -> map -> <k2, v2> -> reduce -> <k3, v3> (output)

二、編寫Map/Reduce代碼

編寫用於WordCount統計詞條的代碼。

000、數據

准備一些文件，用於WordCount統計使用。

我的文件是Hadoop上的：data/123.txt。。。。
內容是一些hadoop hdfs等通過空格分開的詞條

001、依賴包

使用Maven構建項目使用以下依賴即可。

或者自行導入。

<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>${hadoop.version}</version>
</dependency>

002、Map代碼

public class WordMap extends Mapper<LongWritable, Text,Text, IntWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //將序列化的類型轉換為String
        String s = value.toString();
        //工具類
        StringTokenizer tokenizer = new StringTokenizer(s);
        while (tokenizer.hasMoreTokens()){
            context.write(new Text(tokenizer.nextToken()),new IntWritable(1));
        }

    }
}

003、Reduce代碼

public class WordReducer extends Reducer<Text, IntWritable,Text,IntWritable> {

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int count=0;//計數器

        Iterator<IntWritable> iterator = values.iterator();

        while (iterator.hasNext()){
            count+=iterator.next().get();
        }
        System.out.println("reduce over...");
        context.write(key,new IntWritable(count));

    }
}

004、作業代碼

其實很多信息是可以外部輸入的或者寫在配置文件中便於復用，但我直接設置也是可以的。

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

        System.out.println("word count");

        Configuration conf=new Configuration();

        // 指明 HDFS 的地址
        conf.set("fs.defaultFS", "hdfs://192.168.56.101:9000");

        //設置一個用戶，如果當前用戶就是hadoop，則可以略去
        System.setProperty("HADOOP_USER_NAME","hadoop");


        //設置job及相關的各種參數
        Job job = Job.getInstance(conf, "wordcount");
        job.setJarByClass(WordCountDriver.class);
        job.setMapperClass(WordMap.class);
        job.setReducerClass(WordReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //作業的輸入目錄和輸出目錄，輸出目錄必須不存在，hadoop會自動創建
        FileInputFormat.addInputPaths(job,"data");
        FileOutputFormat.setOutputPath(job, new Path("dataDir"));

        //設置job執行等待，並打印進度
        boolean b = job.waitForCompletion(true);

        //作業結束
        System.exit(b?0:1);


    }
}

此時已經可以運行，在ide中運行時會出現相關信息。

此時在命令行下查看結果：

#在相應目錄下生成幾個文件，partxxxx文件是記錄結果的文件
hadoop@Hadoop:~$ hadoop fs -cat /user/hadoop/dataDir/part-r-00000
hadoop	2
hbase	1
hive	1
java	1
spark	2

三、打包執行

上面是在ide下運行的，也經常在服務器中運行，此時需要打包。

由於使用Maven構建，直接打包就行。

mvn clean package

由於信息都是寫死在java文件中的，所以只需要一個main文件名就可以運行，結果也是一樣的。

hadoop jar hadoop-1.0-SNAPSHOT.jar  mapreduce.WordCountDriver