MapReduce的輸出格式是怎樣的

172次閱讀

共計 6345 個字符，預(yù)計需要花費 16 分鐘才能閱讀完成。

本篇內(nèi)容主要講解“MapReduce 的輸出格式是怎樣的”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓丸趣 TV 小編來帶大家學(xué)習(xí)“MapReduce 的輸出格式是怎樣的”吧!

MapReduce 的輸出格式

Hadoop 都有相應(yīng)的輸出格式。默認情況下只有一個 Reduce，輸出只有一個文件，默認文件名為 part-r-00000，輸出文件的個數(shù)與 Reduce 的個數(shù)一致。如果有兩個 Reduce，輸出結(jié)果就有兩個文件，第一個為 part-r-00000，第二個為 part-r-00001，依次類推。

OutputFormat 接口

OutputFormat 主要用于描述輸出數(shù)據(jù)的格式，它能夠?qū)⒂脩籼峁┑?key/value 對寫入特定格式的文件中。通過 OutputFormat 接口，實現(xiàn)具體的輸出格式，過程有些復(fù)雜也沒有這個必要。Hadoop 自帶了很多 OutputFormat 的實現(xiàn)，它們與 InputFormat 實現(xiàn)相對應(yīng)，足夠滿足我們業(yè)務(wù)的需要。OutputFormat 類的層次結(jié)構(gòu)如下圖所示。

OutputFormat 是 MapReduce 輸出的基類，所有實現(xiàn) MapReduce 輸出都實現(xiàn)了 OutputFormat 接口。我們可以把這些實現(xiàn)接口類分為以下幾種類型，分別一一介紹。

文本輸出

默認的輸出格式是 TextOutputFormat，它把每條記錄寫為文本行。它的鍵和值可以是任意類型，因為 TextOutputFormat 調(diào)用 toString() 方法把它們轉(zhuǎn)換為字符串。每個鍵 / 值對由制表符進行分割，當(dāng)然也可以設(shè)定 mapreduce.output.textoutputformat.separator 屬性（舊版本 API 中的 mapred.textoutputformat.separator）改變默認的分隔符。與 TextOutputFormat 對應(yīng)的輸入格式是 KeyValueTextInputFormat，它通過可配置的分隔符將鍵 / 值對文本分割。

可以使用 NullWritable 來省略輸出的鍵或值（或兩者都省略，相當(dāng)于 NullOutputFormat 輸出格式，后者什么也不輸出）。這也會導(dǎo)致無分隔符輸出，以使輸出適合用 TextInputFormat 讀取。

二進制輸出

1、關(guān)于 SequenceFileOutputFormat

顧名思義，SequenceFileOutputFormat 將它的輸出寫為一個順序文件。如果輸出需要作為后續(xù) MapReduce 任務(wù)的輸入，這便是一種好的輸出格式，因為它的格式緊湊，很容易被壓縮。

2、關(guān)于 SequenceFileAsBinaryOutputFormat

SequenceFileAsBinaryOutputFormat 把鍵 / 值對作為二進制格式寫到一個 SequenceFile 容器中。

3、關(guān)于 MapFileOutputFormat

MapFileOutputFormat 把 MapFile 作為輸出。MapFile 中的鍵必須順序添加，所以必須確保 reducer 輸出的鍵已經(jīng)排好序。

多個輸出

上面我們提到，默認情況下只有一個 Reduce，輸出只有一個文件。有時可能需要對輸出的文件名進行控制或讓每個 reducer 輸出多個文件。我們有兩種方式實現(xiàn) reducer 輸出多個文件。

1、Partitioner

我們考慮這樣一個需求：按學(xué)生的年齡段，將數(shù)據(jù)輸出到不同的文件路徑下。這里我們分為三個年齡段：小于等于 20 歲、大于 20 歲小于等于 50 歲和大于 50 歲。

我們采用的方法是每個年齡段對應(yīng)一個 reducer。為此，我們需要通過以下兩步實現(xiàn)。

第一步：把作業(yè)的 reducer 數(shù)設(shè)為年齡段數(shù)即為 3。

job.setPartitionerClass(PCPartitioner.class);// 設(shè)置 Partitioner 類
job.setNumReduceTasks(3);// reduce 個數(shù)設(shè)置為 3

第二步：寫一個 Partitioner，把同一個年齡段的數(shù)據(jù)放到同一個分區(qū)。

public static class PCPartitioner extends Partitioner  Text, Text  {
 @Override
 public int getPartition(Text key, Text value, int numReduceTasks) {
 // TODO Auto-generated method stub
 String[] nameAgeScore = value.toString().split( \t 
 String age = nameAgeScore[1];// 學(xué)生年齡
 int ageInt = Integer.parseInt(age);// 按年齡段分區(qū)
 //  默認指定分區(qū)  0
 if (numReduceTasks == 0)
 return 0;
 // 年齡小于等于 20，指定分區(qū) 0
 if (ageInt  = 20) { return 0;
 //  年齡大于 20，小于等于 50，指定分區(qū) 1
 if (ageInt   20   ageInt  = 50) { return 1 % numReduceTasks;
 //  剩余年齡，指定分區(qū) 2
 else
 return 2 % numReduceTasks;
}

這種方法實現(xiàn)多文件輸出，也只能滿足此種需求。很多情況下是無法實現(xiàn)的，因為這樣做存在兩個缺點。

第一，需要在作業(yè)運行之前需要知道分區(qū)數(shù)和年齡段的個數(shù)，如果分區(qū)數(shù)很大或者未知，就無法操作。

第二，一般來說，讓應(yīng)用程序來嚴(yán)格限定分區(qū)數(shù)并不好，因為可能導(dǎo)致分區(qū)數(shù)少或分區(qū)不均。

2、MultipleOutputs 類

MultipleOutputs 類可以將數(shù)據(jù)寫到多個文件，這些文件的名稱源于輸出的鍵和值或者任意字符串。這允許每個 reducer（或者只有 map 作業(yè)的 mapper）創(chuàng)建多個文件。采用 name-m-nnnnn 形式的文件名用于 map 輸出，name-r-nnnnn 形式的文件名用于 reduce 輸出，其中 name 是由程序設(shè)定的任意名字，nnnnn 是一個指明塊號的整數(shù)（從 0 開始）。塊號保證從不同塊（mapper 或 reducer）寫的輸出在相同名字情況下不會沖突。

假如這里有一份郵箱數(shù)據(jù)文件，我們期望統(tǒng)計郵箱出現(xiàn)次數(shù)并按照郵箱的類別，將這些郵箱分別輸出到不同文件路徑下。數(shù)據(jù)集示例如下所示。

wolys@21cn.com
zss1984@126.com
294522652@qq.com
simulateboy@163.com
zhoushigang_123@163.com
sirenxing424@126.com
lixinyu23@qq.com
chenlei1201@gmail.com
370433835@qq.com
cxx0409@126.com
viv093@sina.com
q62148830@163.com
65993266@qq.com
summeredison@sohu.com
zhangbao-autumn@163.com
diduo_007@yahoo.com.cn
fxh852@163.com

下面我們編寫 MapReduce 程序，實現(xiàn)上述業(yè)務(wù)需求。

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class Email extends Configured implements Tool {
 public static class MailMapper extends Mapper  LongWritable, Text, Text, IntWritable  {private final static IntWritable one = new IntWritable(1);
 @Override
 protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {context.write(value, one);
 public static class MailReducer extends Reducer  Text, IntWritable, Text, IntWritable  {private IntWritable result = new IntWritable();
 private MultipleOutputs  Text, IntWritable  multipleOutputs;
 @Override
 protected void setup(Context context) throws IOException ,InterruptedException{multipleOutputs = new MultipleOutputs  Text, IntWritable (context);
 protected void reduce(Text Key, Iterable  IntWritable  Values,Context context) throws IOException, InterruptedException {int begin = Key.toString().indexOf( @ 
 int end = Key.toString().indexOf( . 
 if(begin =end){
  return;
 }
 // 獲取郵箱類別，比如  qq
 String name = Key.toString().substring(begin+1, end);
 int sum = 0;
 for (IntWritable value : Values) {sum += value.get();
 result.set(sum);
 multipleOutputs.write(Key, result, name);
 @Override
 protected void cleanup(Context context) throws IOException ,InterruptedException{multipleOutputs.close();
 @Override
 public int run(String[] args) throws Exception {Configuration conf = new Configuration();//  讀取配置文件
 Path mypath = new Path(args[1]);
 FileSystem hdfs = mypath.getFileSystem(conf);// 創(chuàng)建輸出路徑
 if (hdfs.isDirectory(mypath)) {hdfs.delete(mypath, true);
 Job job = Job.getInstance();//  新建一個任務(wù)
 job.setJarByClass(Email.class);//  主類
 FileInputFormat.addInputPath(job, new Path(args[0]));//  輸入路徑
 FileOutputFormat.setOutputPath(job, new Path(args[1]));//  輸出路徑
 job.setMapperClass(MailMapper.class);// Mapper
 job.setReducerClass(MailReducer.class);// Reducer
 job.setOutputKeyClass(Text.class);// key 輸出類型
 job.setOutputValueClass(IntWritable.class);// value 輸出類型
 job.waitForCompletion(true);
 return 0;
 public static void main(String[] args) throws Exception {String[] args0 = {
 hdfs://single.hadoop.dajiangtai.com:9000/junior/mail.txt ,
 hdfs://single.hadoop.dajiangtai.com:9000/junior/mail-out/  };
 int ec = ToolRunner.run(new Configuration(), new Email(), args0);
 System.exit(ec);
}

在 reducer 中，在 setup() 方法中構(gòu)造一個 MultipleOutputs 的實例并將它賦給一個實例變量。在 reduce() 方法中使用 MultipleOutputs 實例來寫輸出，而不是 context。write() 方法作用于鍵、值、和名字。

程序運行之后，輸出文件的命名如下所示。

/mail-out/163-r-00000
/mail-out/126-r-00000
/mail-out/21cn-r-00000
/mail-out/gmail-r-00000
/mail-out/qq-r-00000
/mail-out/sina-r-00000
/mail-out/sohu-r-00000
/mail-out/yahoo-r-00000
/mail-out/part-r-00000

在 MultipleOutputs 的 write() 方法中指定的基本路徑相當(dāng)于輸出路徑進行解釋，因為它可以包含文件路徑分隔符 (/)，創(chuàng)建任意深度的子目錄是有可能的。

數(shù)據(jù)庫輸出

DBOutputFormat 適用于將作業(yè)輸出數(shù)據(jù)（中等規(guī)模的數(shù)據(jù)）轉(zhuǎn)存到 Mysql、Oracle 等數(shù)據(jù)庫。

到此，相信大家對“MapReduce 的輸出格式是怎樣的”有了更深的了解，不妨來實際操作一番吧！這里是丸趣 TV 網(wǎng)站，更多相關(guān)內(nèi)容可以進入相關(guān)頻道進行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！

正文完