(转)MultipleOutputFormat和MultipleOutputs

MultipleOutputFormat和MultipleOutputs

http://www.cnblogs.com/liangzh/archive/2012/05/22/2512264.html

一,介绍

1,旧API中有 org.apache.hadoop.mapred.lib.MultipleOutputFormat和org.apache.hadoop.mapred.lib.MultipleOutputs

MultipleOutputFormat allowing to write the output data to different output files.

MultipleOutputs creates multiple OutputCollectors. Each OutputCollector can have its own OutputFormat and types for the key/value pair. Your MapReduce program will decide what to output to each OutputCollector.

2,新API中  org.apache.hadoop.mapreduce.lib.output.MultipleOutputs

整合了上面旧API两个的功能,没有了MultipleOutputFormat。

  The MultipleOutputs class simplifies writing output data to multiple outputs

  Case one: writing to additional outputs other than the job default output. Each additional output, or named output, may be configured with its own             OutputFormat, with its own key class and with its own value class.

  Case two: to write data to different files provided by user

下面这段话来自Hadoop:The.Definitive.Guide(3rd,Early.Release)P251

  “In the old MapReduce API there are two classes for producing multiple outputs: MultipleOutputFormat and MultipleOutputs. In a nutshell, MultipleOutputs is more fully featured, but MultipleOutputFormat has more control over the output directory structure and file naming. MultipleOutputs in the new API combines the best features of the two multiple output classes in the old API.”

二,应用

1,输出到多个文件或多个文件夹:

  驱动中不需要额外改变,只需要在MapClass或Reduce类中加入如下代码

  private MultipleOutputs<Text,IntWritable> mos;
  public void setup(Context context) throws IOException,InterruptedException {
    mos = new MultipleOutputs(context);
  }
  public void cleanup(Context context) throws IOException,InterruptedException {
    mos.close();
  }
  然后就可以用mos.write(Key key,Value value,String baseOutputPath)代替context.write(key, value);
  在MapClass或Reduce中使用,输出时也会有默认的文件part-m-00*或part-r-00*,不过这些文件是无内容的,大小为0. 而且只有part-m-00*会传给Reduce。

2,以多种格式输出:

public class TestwithMultipleOutputs extends Configured implements Tool {

  public static class MapClass extends Mapper<LongWritable,Text,Text,IntWritable> {

    private MultipleOutputs<Text,IntWritable> mos;

    protected void setup(Context context) throws IOException,InterruptedException {
      mos = new MultipleOutputs<Text,IntWritable>(context);
    }

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
      String line = value.toString();
      String[] tokens = line.split("-");

      mos.write("MOSInt",new Text(tokens[0]), new IntWritable(Integer.parseInt(tokens[1])));  //(第一处)
      mos.write("MOSText", new Text(tokens[0]),tokens[2]);     //(第二处)
      mos.write("MOSText", new Text(tokens[0]),line,tokens[0]+"/");  //(第三处)同时也可写到指定的文件或文件夹中
    }

    protected void cleanup(Context context) throws IOException,InterruptedException {
      mos.close();
    }

  }
  public int run(String[] args) throws Exception {

    Configuration conf = getConf();

    Job job = new Job(conf,"word count with MultipleOutputs");

    job.setJarByClass(TestwithMultipleOutputs.class);

    Path in = new Path(args[0]);
    Path out = new Path(args[1]);

    FileInputFormat.setInputPaths(job, in);
    FileOutputFormat.setOutputPath(job, out);

    job.setMapperClass(MapClass.class);
    job.setNumReduceTasks(0);  

    MultipleOutputs.addNamedOutput(job,"MOSInt",TextOutputFormat.class,Text.class,IntWritable.class);
    MultipleOutputs.addNamedOutput(job,"MOSText",TextOutputFormat.class,Text.class,Text.class);

    System.exit(job.waitForCompletion(true)?0:1);
    return 0;
  }

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

    int res = ToolRunner.run(new Configuration(), new TestwithMultipleOutputs(), args);
    System.exit(res); 
  }

}

测试的数据:

abc-1232-hdf
abc-123-rtd
ioj-234-grjth
ntg-653-sdgfvd
kju-876-btyun
bhm-530-bhyt
hfter-45642-bhgf
bgrfg-8956-fmgh
jnhdf-8734-adfbgf
ntg-68763-nfhsdf
ntg-98634-dehuy
hfter-84567-drhuk

结果截图:(结果输出到/test/testMOSout)

遇到的一个问题:

  如果没有mos.close(), 程序运行中会出现异常:

  12/05/21 20:12:47 WARN hdfs.DFSClient: DataStreamer Exception:

  org.apache.hadoop.ipc.RemoteException:org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException: No lease on
  /test/mosreduce/_temporary/_attempt_local_0001_r_000000_0/h-r-00000 File does not exist. [Lease. Holder: DFSClient_-352105532, pendingcreates: 5]

分类:

时间: 2024-08-08 21:49:19

(转)MultipleOutputFormat和MultipleOutputs的相关文章

Hadoop之——MapReduce实战(一)

转载请注明出处:http://blog.csdn.net/l1028386804/article/details/45956487 MapReduce概述      MapReduce是一种分布式计算模型,由Google提出,主要用于搜索领域,解决海量数据的计算问题. MR由两个阶段组成:Map和Reduce,用户只需要实现map()和reduce()两个函数,即可实现分布式计算,非常简单. 这两个函数的形参是key.value对,表示函数的输入信息. MR执行流程 MapReduce原理 执行

MapReduce输入输出类型、格式及实例

输入格式 1.输入分片与记录 2.文件输入 3.文本输入 4.二进制输入 5.多文件输入 6.数据库格式输入 1.输入分片与记录 1.JobClient通过指定的输入文件的格式来生成数据分片InputSplit. 2.一个分片不是数据本身,而是可分片数据的引用. 3.InputFormat接口负责生成分片. InputFormat 负责处理MR的输入部分,有三个作用: 验证作业的输入是否规范. 把输入文件切分成InputSplit. 提供RecordReader 的实现类,把InputSplit

自定义实现InputFormat、OutputFormat

一:自定义实现InputFormat *数据源来自于内存 *1.InputFormat是用于处理各种数据源的,下面是实现InputFormat,数据源是来自于内存. *1.1 在程序的job.setInputFormatClass(MyselfmemoryInputFormat.class); *1.2 实现InputFormat,extends InputFormat< , >,实现其中的两个方法,分别是getSplits(..),createRecordReader(..). *1.3 g

使用hadoop multipleOutputs对输出结果进行不一样的组织

MapReduce job中,可以使用FileInputFormat和FileOutputFormat来对输入路径和输出路径来进行设置.在输出目录中,框架自己会自动对输出文件进行命名和组织,如:part-(m|r)-00000之类.但有时为了后续流程的方便,我们常需要对输出结果进行一定的分类和组织.以前常用的方法是在MR job运行过后,用脚本对目录下的数据进行一次重新组织,变成我们需要的格式.研究了一下MR框架中的MultipleOutputs(是2.0之后的新API,是对老版本中Multip

MultipleOutputs

package MRNB_V4; import java.io.IOException; import java.util.Iterator; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apac

MapReduce处理输出多文件格式(MultipleOutputs)

MultiPleOutputs原理 MapReduce job中,可以使用FileInputFormat和FileOutputFormat来对输入路径和输出路径来进行设置.在输出目录中,框架自己会自动对输出文件进行命名和组织,如part-(m|r)-00000之类,但有时为了后续流程的方便,我们常需要对输出结果进行一定的分类和组织.以前常用的方法是在MR job运行之后,用脚本对目录下的数据进行一次重新组织,变成我们需要的格式. 研究了一下MR框架中的MultipleOutputs(是2.0之后

Hadoop中的MultipleOutputs实践

本例子采用hadoop1.1.2版本 采用气象数据作为处理数据 1.MultipleOutputs例子,具体解释在代码中有注释 package StationPatitioner; import java.io.IOException; import java.util.Iterator; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.

multipleOutputs Hadoop

package org.lukey.hadoop.muloutput; import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io

MapReduce 编程 系列六 MultipleOutputs使用

在前面的例子中,输出文件名是默认的: _logs part-r-00001 part-r-00003 part-r-00005 part-r-00007 part-r-00009 part-r-00011 part-r-00013 _SUCCESS part-r-00000 part-r-00002 part-r-00004 part-r-00006 part-r-00008 part-r-00010 part-r-00012 part-r-00014 part-r-0000N 还有一个_SUC