Hadoop之WordCount詳解

花了好長時間查找資料理解、學習、總結 這應該是一篇比較全面的MapReduce之WordCount文章了 耐心看下去java

1,建立本地文件

在hadoop-2.6.0文件夾下建立一個文件夾data,在其中建立一個text文件 node

mkdir data
cd data
vi hello

再在當前文件夾中建立一個apps文件夾,方便後續傳jar包git

mkdir apps

將文本文件傳到HDFS的根目錄下程序員

bin/hdfs dfs -put data/hello /

2,程序打jar包並上傳到apps目錄

3,執行Hadoop命令

bin/hadoop jar apps/WordClass-***.jar /hello /out

4,查看輸出結果

將HDFS根目錄下的/out輸出文件傳到本地目錄中查看,一般有兩個文件:github

5,WordCount程序詳解

這部分是最重要的,可是也是最容易讓人犯暈的部分,涉及到許多mapreduce的原理,可是學習就是這樣,你越難吃透的東西,一般越重要apache

先把程序貼上來:編程

package cn.hx.test;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;

public class WordCountApp {
    //自定義的mapper,繼承org.apache.hadoop.mapreduce.Mapper
    public static class MyMapper extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, LongWritable>{
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context)
                throws IOException, InterruptedException {
            String line = value.toString();
//split  函數是用於按指定字符(串)或正則去分割某個字符串,結果以字符串數組形式返回,這裏按照"\t"來分割text文件中字符,即一個製表符,這就是爲何我在文本中用了空格分割,致使最後的結果有很大的出入。 String[] splited
= line.split("\t");
      //foreach 就是 for(元素類型t 元素變量x:遍歷對象obj){引用x的java語句}
for (String word : splited) { context.write(new Text(word), new LongWritable(1)); } } } public static class MyReducer extends org.apache.hadoop.mapreduce.Reducer<Text, LongWritable, Text, LongWritable>{ @Override protected void reduce(Text k2, Iterable<LongWritable> v2s, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException { long count = 0L; for (LongWritable v2 : v2s) { count += v2.get(); } LongWritable v3 = new LongWritable(count); context.write(k2, v3); } } //客戶端代碼,寫完交給ResourceManager框架去執行 public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, WordCountApp.class.getSimpleName()); //打成jar執行 job.setJarByClass(WordCountApp.class); //數據在哪裏? FileInputFormat.setInputPaths(job, args[0]); //使用哪一個mapper處理輸入的數據? job.setMapperClass(MyMapper.class); //map輸出的數據類型是什麼? job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(LongWritable.class); //使用哪一個reducer處理輸入的數據? job.setReducerClass(MyReducer.class); //reduce輸出的數據類型是什麼? job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); //數據輸出到哪裏? FileOutputFormat.setOutputPath(job, new Path(args[1])); //交給yarn去執行,直到執行結束才退出本程序 job.waitForCompletion(true); } }

 

 POM文件:數組

 

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.hx.test</groupId>
    <artifactId>WordClass</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>WordCount</name>
    <url>http://maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
    </properties>

    <build>
        <pluginManagement>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-jar-plugin</artifactId>
                    <configuration>

                        <archive>
                            <manifest>
                                <mainClass>cn.hx.test.WordClass</mainClass>
                                <addClasspath>true</addClasspath>
                                <classpathPrefix>lib/</classpathPrefix>
                            </manifest>

                        </archive>
                        <classesDirectory>
                        </classesDirectory>
                    </configuration>
                </plugin>
            </plugins>
        </pluginManagement>
    </build>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.6.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.6.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.6.0</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
    </dependencies>
</project>

 

 

如今來一部分 一部分的理解程序:緩存

要寫一個mapreduce程序,首先要實現一個map函數和reduce函數。咱們看看map的方法:app

 

protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context)

**//** * KEYIN 即K1 表示每一行的起始位置(偏移量offset) * VALUEIN 即v1 表示每一行的文本內容 * KEYOUT 即k2 表示每一行中的每一個單詞 * VALUEOUT 即v2 表示每一行中的每一個單詞的出現次數,固定值1

這裏有三個參數,前面兩個LongWritable key, Text value就是輸入的key和value,第三個參數Context context這是能夠記錄輸入的key和value,例如:

context.write(new Text(word), new LongWritable(1));

此外context還會記錄map運算的狀態。

對於reduce函數的方法:

 MapReduce的運行機制:

首先是客戶端要編寫好mapreduce程序,配置好mapreduce的做業也就是job,接下來就是提交job了,提交job是提交到JobTracker上的,這個時候JobTracker就會構建這個job,具體就是分配一個新的job任務的ID值,接下來它會作檢查操做,這個檢查就是肯定輸出目錄是否存在,若是存在那麼job就不能正常運行下去,JobTracker會拋出錯誤給客戶端,接下來還要檢查輸入目錄是否存在,若是不存在一樣拋出錯誤,若是存在JobTracker會根據輸入計算輸入分片(Input Split),若是分片計算不出來也會拋出錯誤,至於輸入分片後面會作講解的,這些都作好了JobTracker就會配置Job須要的資源了。分配好資源後,JobTracker就會初始化做業,初始化主要作的是將Job放入一個內部的隊列,讓配置好的做業調度器能調度到這個做業,做業調度器會初始化這個job,初始化就是建立一個正在運行的job對象(封裝任務和記錄信息),以便JobTracker跟蹤job的狀態和進程。

初始化完畢後,做業調度器會獲取輸入分片信息(input split),每一個分片建立一個map任務。接下來就是任務分配了,這個時候tasktracker會運行一個簡單的循環機制按期發送心跳給jobtracker,心跳間隔是5秒,程序員能夠配置這個時間,心跳就是jobtracker和tasktracker溝通的橋樑,經過心跳,jobtracker能夠監控tasktracker是否存活,也能夠獲取tasktracker處理的狀態和問題,同時tasktracker也能夠經過心跳裏的返回值獲取jobtracker給它的操做指令。任務分配好後就是執行任務了。在任務執行時候jobtracker能夠經過心跳機制監控tasktracker的狀態和進度,同時也能計算出整個job的狀態和進度,而tasktracker也能夠本地監控本身的狀態和進度。當jobtracker得到了最後一個完成指定任務的tasktracker操做成功的通知時候,jobtracker會把整個job狀態置爲成功,而後當客戶端查詢job運行狀態時候(注意:這個是異步操做),客戶端會查到job完成的通知的。若是job中途失敗,mapreduce也會有相應機制處理,通常而言若是不是程序員程序自己有bug,mapreduce錯誤處理機制都能保證提交的job能正常完成。

下面我從邏輯實體的角度講解mapreduce運行機制,這些按照時間順序包括:輸入分片(input split)、map階段、combiner階段、shuffle階段和reduce階段

1. 輸入分片(input split):在進行map計算以前,mapreduce會根據輸入文件計算輸入分片(input split),每一個輸入分片(input split)針對一個map任務,輸入分片(input split)存儲的並不是數據自己,而是一個分片長度和一個記錄數據的位置的數組,輸入分片(input split)每每和hdfs的block(塊)關係很密切,假如咱們設定hdfs的塊的大小是64mb,若是咱們輸入有三個文件,大小分別是3mb、65mb和127mb,那麼mapreduce會把3mb文件分爲一個輸入分片(input split),65mb則是兩個輸入分片(input split)而127mb也是兩個輸入分片(input split),換句話說咱們若是在map計算前作輸入分片調整,例如合併小文件,那麼就會有5個map任務將執行,並且每一個map執行的數據大小不均,這個也是mapreduce優化計算的一個關鍵點。

2. map階段:就是程序員編寫好的map函數了,所以map函數效率相對好控制,並且通常map操做都是本地化操做也就是在數據存儲節點上進行;

3. combiner階段:combiner階段是程序員能夠選擇的,combiner其實也是一種reduce操做,所以咱們看見WordCount類裏是用reduce進行加載的。Combiner是一個本地化的reduce操做,它是map運算的後續操做,主要是在map計算出中間文件前作一個簡單的合併重複key值的操做,例如咱們對文件裏的單詞頻率作統計,map計算時候若是碰到一個hadoop的單詞就會記錄爲1,可是這篇文章裏hadoop可能會出現n屢次,那麼map輸出文件冗餘就會不少,所以在reduce計算前對相同的key作一個合併操做,那麼文件會變小,這樣就提升了寬帶的傳輸效率,畢竟hadoop計算力寬帶資源每每是計算的瓶頸也是最爲寶貴的資源,可是combiner操做是有風險的,使用它的原則是combiner的輸入不會影響到reduce計算的最終輸入,例如:若是計算只是求總數,最大值,最小值可使用combiner,可是作平均值計算使用combiner的話,最終的reduce計算結果就會出錯。

4. shuffle階段:將map的輸出做爲reduce的輸入的過程就是shuffle了,這個是mapreduce優化的重點地方。這裏我不講怎麼優化shuffle階段,講講shuffle階段的原理,由於大部分的書籍裏都沒講清楚shuffle階段。Shuffle一開始就是map階段作輸出操做,通常mapreduce計算的都是海量數據,map輸出時候不可能把全部文件都放到內存操做,所以map寫入磁盤的過程十分的複雜,更況且map輸出時候要對結果進行排序,內存開銷是很大的,map在作輸出時候會在內存裏開啓一個環形內存緩衝區,這個緩衝區專門用來輸出的,默認大小是100mb,而且在配置文件裏爲這個緩衝區設定了一個閥值,默認是0.80(這個大小和閥值都是能夠在配置文件裏進行配置的),同時map還會爲輸出操做啓動一個守護線程,若是緩衝區的內存達到了閥值的80%時候,這個守護線程就會把內容寫到磁盤上,這個過程叫spill,另外的20%內存能夠繼續寫入要寫進磁盤的數據,寫入磁盤和寫入內存操做是互不干擾的,若是緩存區被撐滿了,那麼map就會阻塞寫入內存的操做,讓寫入磁盤操做完成後再繼續執行寫入內存操做,前面我講到寫入磁盤前會有個排序操做,這個是在寫入磁盤操做時候進行,不是在寫入內存時候進行的,若是咱們定義了combiner函數,那麼排序前還會執行combiner操做。

每次spill操做也就是寫入磁盤操做時候就會寫一個溢出文件,也就是說在作map輸出有幾回spill就會產生多少個溢出文件,等map輸出所有作完後,map會合並這些輸出文件。這個過程裏還會有一個Partitioner操做,對於這個操做不少人都很迷糊,其實Partitioner操做和map階段的輸入分片(Input split)很像,一個Partitioner對應一個reduce做業,若是咱們mapreduce操做只有一個reduce操做,那麼Partitioner就只有一個,若是咱們有多個reduce操做,那麼Partitioner對應的就會有多個,Partitioner所以就是reduce的輸入分片,這個程序員能夠編程控制,主要是根據實際key和value的值,根據實際業務類型或者爲了更好的reduce負載均衡要求進行,這是提升reduce效率的一個關鍵所在。到了reduce階段就是合併map輸出文件了,Partitioner會找到對應的map輸出文件,而後進行復制操做,複製操做時reduce會開啓幾個複製線程,這些線程默認個數是5個,程序員也能夠在配置文件更改複製線程的個數,這個複製過程和map寫入磁盤過程相似,也有閥值和內存大小,閥值同樣能夠在配置文件裏配置,而內存大小是直接使用reduce的tasktracker的內存大小,複製時候reduce還會進行排序操做和合並文件操做,這些操做完了就會進行reduce計算了。

5. reduce階段:和map函數同樣也是程序員編寫的,最終結果是存儲在hdfs上的。

Mapreduce的相關問題

① jobtracker的單點故障:jobtracker和hdfs的namenode同樣也存在單點故障,單點故障一直是hadoop被人詬病的大問題,爲何hadoop的作的文件系統和mapreduce計算框架都是高容錯的,可是最重要的管理節點的故障機制卻如此很差,我認爲主要是namenode和jobtracker在實際運行中都是在內存操做,而作到內存的容錯就比較複雜了,只有當內存數據被持久化後容錯纔好作,namenode和jobtracker均可以備份本身持久化的文件,可是這個持久化都會有延遲,所以真的出故障,任然不能總體恢復,另外hadoop框架裏包含zookeeper框架,zookeeper能夠結合jobtracker,用幾臺機器同時部署jobtracker,保證一臺出故障,有一臺立刻能補充上,不過這種方式也無法恢復正在跑的mapreduce任務。

② 作mapreduce計算時候,輸出通常是一個文件夾,並且該文件夾是不能存在,並且這個檢查作的很早,當咱們提交job時候就會進行,mapreduce之因此這麼設計是保證數據可靠性,若是輸出目錄存在reduce就搞不清楚你究竟是要追加仍是覆蓋,不論是追加和覆蓋操做都會有可能致使最終結果出問題,mapreduce是作海量數據計算,一個生產計算的成本很高,例如一個job徹底執行完可能要幾個小時,所以一切影響錯誤的狀況mapreduce是零容忍的。

③ Mapreduce還有一個InputFormat和OutputFormat,咱們在編寫map函數時候發現map方法的參數是之間操做行數據,沒有牽涉到InputFormat,這些事情在咱們new Path時候mapreduce計算框架幫咱們作好了,而OutputFormat也是reduce幫咱們作好了,咱們使用什麼樣的輸入文件,就要調用什麼樣的InputFormat,InputFormat是和咱們輸入的文件類型相關的,mapreduce裏經常使用的InputFormat有FileInputFormat普通文本文件,SequenceFileInputFormat是指hadoop的序列化文件,另外還有KeyValueTextInputFormat。OutputFormat就是咱們想最終存儲到hdfs系統上的文件格式了,這個根據你須要定義了,hadoop有支持不少文件格式。

相關文章
相關標籤/搜索