2. 程式人生 > >MapReduce輸入輸出型別、格式及例項

MapReduce輸入輸出型別、格式及例項

阿新 • • 發佈：2019-01-26

輸入格式

1、輸入分片與記錄
2、檔案輸入
3、文字輸入
4、二進位制輸入
5、多檔案輸入
6、資料庫格式輸入

1、輸入分片與記錄

1、JobClient通過指定的輸入檔案的格式來生成資料分片InputSplit。
2、一個分片不是資料本身，而是可分片資料的引用。
3、InputFormat介面負責生成分片。

InputFormat 負責處理MR的輸入部分，有三個作用:
驗證作業的輸入是否規範。
把輸入檔案切分成InputSplit。
提供RecordReader 的實現類，把InputSplit讀到Mapper中進行處理。

2、檔案輸入

抽象類：FilelnputFormat


1、FilelnputFormat是所有使用檔案作為資料來源的InputFormat實現的基類。
2、FilelnputFormat輸入資料格式的分片大小由資料塊大小決定。

FileInputFormat儲存作為job輸入的所有檔案，並實現了對輸入檔案計算splits的方法。至於獲得記錄的方法是有不同的子類——TextInputFormat進行實現的。

package org.apache.hadoop.mapreduce.lib.input;
public abstract class FileInputFormat<K, V> extends InputFormat<K, V> {
  protected
 
 long computeSplitSize(long blockSize, long minSize,long maxSize) {
    return Math.max(minSize, Math.min(maxSize, blockSize));
  }

  /*Generate the list of files and make them into FileSplits.*/
  public List<InputSplit> getSplits(JobContext job) throws IOException {
     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
     long
 
 maxSize = getMaxSplitSize(job);
     ......
     long blockSize = file.getBlockSize();
     long splitSize = computeSplitSize(blockSize, minSize, maxSize);
     ......
  }
  /*Get the minimum split size*/
  public static long getMinSplitSize(JobContext job) {
    return job.getConfiguration().getLong(SPLIT_MINSIZE, 1L);
  }

  /*Get the maximum split size.*/
  public static long getMaxSplitSize(JobContext context) {
    return context.getConfiguration().getLong(SPLIT_MAXSIZE,Long.MAX_VALUE);
  }

    //是否分片
    /*
    Is the given filename splitable? Usually, true, but if the file is stream compressed, it will not be.
    <code>FileInputFormat</code> implementations can override this and return <code>false</code> to ensure that individual input files are never split-up so that {@link Mapper}s process entire files.
    */
    protected boolean isSplitable(JobContext context, Path filename) {
    return true;//預設需要分片
  }

}
自定義輸入格式

如果我們不需要分片，那我們就需要對isSplitable方法進行重寫
1、繼承FileInputFormat基類。
2、重寫裡面的getSplits(JobContext context)方法。
3、重寫createRecordReader(InputSplit split,TaskAttemptContext context)方法。
詳細例子：
http://blog.csdn.net/scgaliguodong123_/article/details/46492039

InputSplit

在執行mapreduce之前，原始資料被分割成若干split，每個split作為一個map任務的輸入，在map執行過程中split會被分解成一個個記錄（key-value對）， map會依次處理每一個記錄。
FileInputFormat只劃分比HDFS block大的檔案，所以FileInputFormat劃分
的結果是這個檔案或者是這個檔案中的一部分。
如果一個檔案的大小比block小，將不會被劃分，這也是Hadoop處理大檔案
的效率要比處理很多小檔案的效率高的原因。
當Hadoop處理很多小檔案（檔案大小小於hdfs block大小）的時候，由於
FileInputFormat不會對小檔案進行劃分，所以每一個小檔案都會被當做一個split並分配一個map任務，導致效率底下。
例如：一個1G的檔案，會被劃分成16個64MB的split，並分配16個map任務處
理，而10000個100kb的檔案會被10000個map任務處理。

Map任務的數量？

一個InputSplit對應一個Map task。
InputSplit的大小是由Math.max(minSize,Math.min(maxSize, blockSize))決定。
單節點一般10-100個map task。map task執行時長不建議低於1 分鐘，否
則效率低。

抽象類：CombineFilelnputFormat
1、可以使用CombineFilelnputFormat來合併小檔案。

2、因為CombineFilelnputFormat是一個抽象類，使用的時候需要建立一個
CombineFilelnputFormat的實體類，並且實現getRecordReader()的方法。

3、避免檔案分法的方法：
A、資料塊大小盡可能大，這樣使檔案的大小小於資料塊的大小，就不用進行分片。（這種方式不太友好）
B、繼承FilelnputFormat，並且重寫isSplitable(）方法。

job.setInputFormatClass(CombineTextInputFormat.class);

Hadoop2.6.0 CombineTextInputFormat原始碼：

package org.apache.hadoop.mapreduce.lib.input;
/* Input format that is a <code>CombineFileInputFormat</code>-equivalent for <code>TextInputFormat</code>.*/
public class CombineTextInputFormat
  extends CombineFileInputFormat<LongWritable,Text> {

  public RecordReader<LongWritable,Text> createRecordReader(InputSplit split,
    TaskAttemptContext context) throws IOException {
    return new CombineFileRecordReader<LongWritable,Text>(
      (CombineFileSplit)split, context, TextRecordReaderWrapper.class);
  }

  /*A record reader that may be passed to <code>CombineFileRecordReader</code> so that it can be used in a <code>CombineFileInputFormat</code>-equivalent for <code>TextInputFormat</code>.*/
  private static class TextRecordReaderWrapper
    extends CombineFileRecordReaderWrapper<LongWritable,Text> {
    // this constructor signature is required by CombineFileRecordReader
    public TextRecordReaderWrapper(CombineFileSplit split,
      TaskAttemptContext context, Integer idx)
      throws IOException, InterruptedException {
      super(new TextInputFormat(), split, context, idx);
    }
  }
}

3、文字輸入

類名：TextlnputFormat
1、TextlnputFormat是預設的lnputFormat，每一行資料就是一條記錄。

2、TextlnputFormat的key是LongWritable型別的，儲存該行在整個檔案的偏移量，value是每行的資料內容，Text型別。

3、輸入分片與HDFS資料塊關係：TextlnputFormat每一條記錄就是一行，很有可能某一行跨資料塊存放。預設以\n或回車鍵作為一行記錄。

4、TextInputFormat繼承了FileInputFormat。

類名：KeyValueTextInputFormat
可以通過設定key為行號的方式來知道記錄的行號，並且可以通過key.value.separator.in.input設定key與value的分割符。
當輸入資料的每一行是兩列，並用tab分離的形式的時候，KeyValueTextInputformat處理這種格式的檔案非常適合。

如果行中有分隔符，那麼分隔符前面的作為key，後面的作為value；如果行中沒有分隔符，那麼整行作為key，value為空。

job.setInputFormatClass(KeyValueTextInputFormat.class);
//預設分隔符就是製表符
//conf.setStrings(KeyValueLineRecordReader.KEY_VALUE
_SEPERATOR, "\t")

類名：NLineInputFormat
可以設定每個mapper處理的行數，可以通過mapred.line.input.format.lienspermap屬性設定。
NLineInputformat可以控制在每個split中資料的行數。

//設定具體輸入處理類
job.setInputFormatClass(NLineInputFormat.class);
//設定每個split的行數
NLineInputFormat.setNumLinesPerSplit(job, Integer.parseInt(args[2]));

4、二進位制輸入

輸入類：

SequenceFileInputFormat 將key和value以sequencefile格式輸入。
SequenceFileAsTextInputFormat 
SequenceFileAsBinaryInputFormat 將key和value以原始二進位制的格式輸入。

由於SequenceFile能夠支援Splittable，所以能夠作為mapreduce輸入檔案的格式，能夠很方便的得到己經含有＜key,value＞的分片。

SequenceFile處理、壓縮處理。

5、多檔案輸入

類名：MultipleInputs
1、MultipleInputs能夠提供多個輸入資料型別。
2、通過addInputPath()方法來設定多路徑。

6、資料庫格式輸入

類名：DBInputFormat
1、DBInputFormat是一個使用JDBC方式連線資料庫，並且從關係型資料庫中讀取資料的一種輸入格式。
2、有多個map會去連線資料庫，有可能造成資料庫崩潰，因此，避免過多的資料庫連線。
3、HBase中的TablelnputFormat可以讓MapReduce程式訪問HBase表裡的資料。

例項單輸入路徑

[root@master liguodong]# hdfs dfs -cat /input.txt
hello you
hello everybody
hello hadoop
[root@master liguodong]# hdfs dfs -text /tmp.seq
15/06/10 21:17:11 INFO bzip2.Bzip2Factory: Successfully loaded & initialized native-bzip2 library system-native
15/06/10 21:17:11 INFO compress.CodecPool: Got brand-new decompressor [.bz2]
100     apache software
99      chinese good
98      james NBA
97      index pass
96      apache software
95      chinese good
94      james NBA
93      index pass
......
package mrinputformat;

import java.io.IOException;
import java.util.StringTokenizer;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class TestInputFormat {

    public static class TokenizerMapper
       extends Mapper<IntWritable, Text, Text, IntWritable>{

        private final static IntWritable one = new IntWritable(1);//1
        private Text word = new Text();

        public void map(IntWritable key, Text value, Context context
                        ) throws IOException, InterruptedException 
        {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                //k v
                context.write(word, one);
            }
        }
    }

    public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                       Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) 
            {
                sum += val.get();
            }
            result.set(sum); 
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        //1、配置  
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count"); 

        //2、打包執行必須執行的方法
        job.setJarByClass(TestInputFormat.class);

        //3、輸入路徑  
        //hdfs://master:8020/tmp.seq
        //hdfs://master:8020/output
        FileInputFormat.addInputPath(job, new Path(args[0]));  
        //預設是TextInputFormat
        job.setInputFormatClass(SequenceFileInputFormat.class);

        //4、Map
        job.setMapperClass(TokenizerMapper.class);

        //5、Combiner
        job.setCombinerClass(IntSumReducer.class);

        //6、Reducer
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);

        //7、 輸出路徑
        FileOutputFormat.setOutputPath(job, new Path(args[1])); 

        //8、提交作業
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

執行結果：

多輸入路徑方式

package mrinputformat;

import java.io.IOException;
import java.util.StringTokenizer;

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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class TestInputFormat {

    //採用TextInputFormat
    public static class Mapper1
       extends Mapper<LongWritable, Text, Text, IntWritable>{

        private final static IntWritable one = new IntWritable(1);//1
        private Text word = new Text();

        public void map(LongWritable key, Text value, Context context
                        ) throws IOException, InterruptedException 
        {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                //k v
                context.write(word, one);
            }
        }
    }

    //SequenceFileInputFormat
    public static class Mapper2
       extends Mapper<IntWritable, Text, Text, IntWritable>{

        private final static IntWritable one = new IntWritable(1);//1
        private Text word = new Text();

        public void map(IntWritable key, Text value, Context context
                        ) throws IOException, InterruptedException 
        {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                //k v
                context.write(word, one);
            }
        }
    }

    public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                       Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) 
            {
                sum += val.get();
            }
            result.set(sum); 
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        //1、配置  
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count"); 

        //2、打包執行必須執行的方法
        job.setJarByClass(TestInputFormat.class);

        //3、輸入路徑  
        //hdfs://master:8020/tmp.seq
        //hdfs://master:8020/output
        //單個輸入路徑
        //FileInputFormat.addInputPath(job, new Path(args[0]));  
        //預設是TextInputFormat
        //job.setInputFormatClass(SequenceFileInputFormat.class);
        //4、Map
        //job.setMapperClass(TokenizerMapper.class);

        //多個輸入路徑
        Path path1 = new Path("hdfs://master:8020/input.txt");
        Path path2 = new Path("hdfs://master:8020/tmp.seq");
        MultipleInputs.addInputPath(job, path1, TextInputFormat.class,Mapper1.class);
        MultipleInputs.addInputPath(job, path2, SequenceFileInputFormat.class,Mapper2.class);

        //5、Combiner
        job.setCombinerClass(IntSumReducer.class);

        //6、Reducer
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //7、 輸出路徑
        FileOutputFormat.setOutputPath(job, new Path("hdfs://master:8020/output"));

        //8、提交作業
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

執行結果：

輸出格式

文字輸出

TextOutputFormat

預設的輸出格式，key是LongWritable，value是Text型別， key和value中間值用tab隔開的。

二進位制輸出

SequenceFileOutputFormat
將key和value以sequencefile格式輸出。

SequenceFileAsBinaryOutputFormat
將key和value以原始二進位制的格式輸出。

MapFileOutputFormat
將key和value寫入MapFile中。由於MapFile中的key是有序的，所以寫入的時候必須保證記錄是按key值順序寫入的。

多檔案輸出

MultipleOutputFormat
MultipleOutputs

預設情況下一個reducer會產生一個輸出，但是有些時候我們想一個reducer產生多個輸出， MultipleOutputFormat和MultipleOutputs可以實現這個功能。
區別：MultipleOutputs可以產生不同型別的輸出。

資料庫格式輸出

DBOutputFormat

相關推薦

MapReduce輸入輸出型別、格式及例項

輸入格式 1、輸入分片與記錄 2、檔案輸入 3、文字輸入 4、二進位制輸入 5、多檔案輸入 6、資料庫格式輸入 1、輸入分片與記錄 1、JobClient通過指定的輸入檔案的格式來生成資料分片InputSplit。 2、一個分片不是資料本身

MapReduce輸入輸出類型、格式及實例

回車 行記錄 ade nts cte ftw result 默認 math 輸入格式 1、輸入分片與記錄 2、文件輸入 3、文本輸入 4、二進制輸入 5、多文件輸入 6、數據庫格式輸入 1、輸入分片與記錄 1、JobClient通過指

Hadoop MapReduce輸入輸出類型

imu finally configure 獲得 命名 pfile 計算 uil 大文件 一、輸入格式 　　1、輸入分片split 　　　　　　一個分片對應一個map任務； 　　　　　　一個分片包含一個表（整個文件）上的若幹行，而一條記錄（單行）對應一行； 　　　　　　分片

JS簡介，JS如何實現輸入輸出語句、註釋

同時 原創 rst 多條語句 筆記 innerhtml first js簡介 字符串 JavaScript 簡介 JavaScript 是世界上最流行的編程語言 可用於 HTML 和 web，更可廣泛用於服務器、PC、筆記本電腦、平板電腦和智能手機等設備 JavaScri

016-Hadoop Hive sql語法詳解6-job輸入輸出優化、數據剪裁、減少job數、動態分區

分享 hive table 取數 nbsp put union 正在 style 一、job輸入輸出優化 善用muti-insert、union all，不同表的union all相當於multiple inputs，同一個表的union all，相當map一次輸出多條

IO流：System.in、SequenceInputStream合並流、內存輸入輸出流、數據流

string類型 釋放 rgs int() 當前 控制臺輸出 exc extend 清空 1、System.in System.out 是常用的在控制臺輸出數據的 System.in 可以從控制臺輸入數據 InputStream is = System.

TSM檔案格式及例項解析（四）——string的排列

目錄 引： 資料 簡析 Indexes: blocks: 關聯string 小結: TSM檔案格式現已基本明瞭，某些壓縮演算法有空再去研究一下。 引： 現在想知道未壓縮的檔案格式是怎樣。 但是翻遍官方配置文件 https://docs.influx

大資料培訓之核心知識點Hbase、Hive、Spark和MapReduce的概念理解、特點及機制等

今天，上海尚學堂大資料培訓班畢業的一位學生去參加易普軟體公司面試，應聘的職位是大資料開發。面試官問了他10個問題，主要集中在Hbase、Spark、Hive和MapReduce上，基礎概念、特點、應用場景等問得多。看來，還是非常注重基礎的牢固。整個大資料開發技術，這幾個技術知識點佔了很大一部分。那本

javaSE (三十三）其他流（序列流、記憶體輸出流、隨機訪問流、物件操作流、資料輸入輸出流、列印流、標準輸入輸出流、properties）

1、序列流（SequenceInputStream ）： 序列流主要的作用就是整合位元組輸入流，將很多的進口整合成一個 這裡著重講一下多於兩個輸入流的整合： 步驟： 建立三個輸入流 建立vector集合存入這些輸入流 將這些輸入流變成列舉型別 Vector.e

Redis學習小結--windows下安裝、基本資料型別、持久化及Jedis

　　Redis：REmote DIctionary Server 遠端字典伺服器 　　　　是一個開源免費，ANSI C編寫，遵守BSD協議，支援網路, Key-Value分散式記憶體DB,基於記憶體執行，支援持久化NoSql DB,資料結構伺服器,提供多種語言的API 一：在Window下安裝

redis的型別、原理及使用場景

轉自：http://blog.csdn.net/hguisu/article/details/8836819 1.  MySQL+Memcached架構的問題 Memcached採用客戶端-伺服器的架構，客戶端和伺服器端的通訊使用自定義的協議標準，只要滿足協議格

C#呼叫C++ 平臺呼叫P/Invoke 結構體--輸入輸出引數、返回值、返出值、結構體陣列作為引數【五】

【1】結構體作為輸入輸出引數 C++程式碼： typedef struct _testStru1 { int iVal; char cVal; __int64 llVal; }testS

深究遞迴和迭代的區別、聯絡、優缺點及例項對比

萬物的存在是需要時間的檢驗的，遞迴沒有被歷史所埋沒，即有存在的理由。從理論上說，所有的遞迴函式都可以轉換為迭代函式，反之亦然，然而代價通常都是比較高的。但從演算法結構來說，遞迴宣告的結構並不總能夠轉換為迭代結構，原因在於結構的引申本身屬於遞迴的概念，用迭代的方法在設計初期根本無法實現，這就像動多型的東西並不總

要求從使用者輸入的多行文字中提取學生的姓名、學號及登入日期，並封裝到 Student 類中作為類的私有屬性。建立一個 Student 型別的物件陣列，對學號進行 升序排序並輸出

/*從鍵盤輸入多行文字，格式如下：學生端名稱,姓名,班級名稱,學生 ID,註冊時間 姜濤,姜濤,,20092212232,2011-11-4 9:06:56 任超,任超,,20092212239,2011-11-4 9:06:56 楊陽,楊陽,,20092212302,20

要求從使用者輸入的多行文字中提取學生的姓名、學號及登入日期，並封裝到 Student 類中作為類的私有屬性。建立一個Student型別的物件陣列，對學號進行 升序排序並輸出。

/*部落格網站設定了校驗密碼的規則，編寫方法檢驗一個字串是否是合法的密碼。規則如下：  密碼長度在8-16之間  密碼只能包含字母和數字  密碼必須存在至少2個數字 如果使用者輸入的密碼符合規則就顯示valid password，否則提示Invalid passwor

MapReduce型別與格式(輸入與輸出)

一、輸入格式 （1）輸入分片記錄 ①JobClient通過指定的輸入檔案的格式來生成資料分片InputSplit； ②一個分片不是資料本身，而是可分片資料的引用； ③InputFormat介面負責生成分片； 原始碼位置：org.apache.hadoop.mapreduce.lib.input包(新)   

Python基礎（Linux下）---python的認識,安裝,資料型別,輸入輸出及分支語句

認識python Python是著名的“龜叔”Guido van Rossum在1989年聖誕節期間，為了打發無聊的聖誕節而編寫的一個程式語言。 Python為我們提供了非常完善的基礎程式碼庫，覆蓋了網路、檔案、GUI、資料庫、文字等大量內容，被形象地稱作“內建電池（batteries

SpringMVC對列舉型別輸入輸出及Mybatis對列舉型別的儲存查詢

文章目錄 摘要 準備環境 Spring MVC接收列舉型別 Spring MVC輸出列舉型別 MyBatis儲存查詢列舉型別 測試 摘要 本文使用SpringBoot.2.0.1.RELEA

【MapReduce詳解及原始碼解析（一）】——分片輸入、Mapper及Map端Shuffle過程

title: 【MapReduce詳解及原始碼解析（一）】——分片輸入、Mapper及Map端Shuffle過程 date: 2018-12-03 21:12:42 tags: Hadoop categories: 大資料 toc: true 點選檢視我的部落格：Josonlee’

C++Primer Plus筆記——第十七章 輸入、輸出和檔案總結及程式清單

目錄 本章小結 程式清單 核心格式化 本章小結        流是進出程式的位元組流。緩衝區足記憶體中的臨時儲存區域，是程式與檔案或其他I/O裝置之間的橋樑。 資訊在緩衝區和檔案之間傳輸時，將使用裝置（如磁碟驅動器）處理效率最高的尺寸