TUNANのBlogHomeArchivesTagsCategories
Hadoop多目录输出
|Hadoop
Word count: 3.3k|Reading time: 15 min

目录

  1. MultipleOutputFormat
  2. MultipleOutputs

MultipleOutputFormat

直到目前，我们看到的所有Mapreduce作业都输出一组文件。但是，在一些场合下，经常要求我们将输出多组文件或者把一个数据集分为多个数据集更为方便；比如将一个log里面属于不同业务线的日志分开来输出，并交给相关的业务线。

用过旧API的人应该知道，旧API中有 org.apache.hadoop.mapred.lib.MultipleOutputFormat和org.apache.hadoop.mapred.lib.MultipleOutputs，文档对MultipleOutputFormat的解释（MultipleOutputs 解释在后面）是：

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

MultipleOutputFormat可以将相似的记录输出到相同的数据集。在写每条记录之前，MultipleOutputFormat将调用generateFileNameForKeyValue方法来确定需要写入的文件名。通常，我们都是继MultipleTextOutputFormat类，来重新实现generateFileNameForKeyValue方法以返回每个输出键/值对的文件名。generateFileNameForKeyValue方法的默认实现如下：

protected String generateFileNameForKeyValue(K key, V value, String name) {
    return name;
}

返回默认的name，我们可以在自己的类中重写这个方法，来定义自己的输出路径，比如：

public static class PartitionFormat
            extends MultipleTextOutputFormat<NullWritable, Text> {
 
        @Override
        protected String generateFileNameForKeyValue(
                NullWritable key,
                Text value, 
                String name) {
            String[] split = value.toString().split(",", -1);
            String country = split[4].substring(1, 3);
            return country + "/" + name;
        }
}

这样相同country的记录将会输出到同一目录下的name文件中。完整的例子如下：

package com.wyp;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
 
import java.io.IOException;
 

public class OutputTest {
    public static class MapClass extends MapReduceBase 
            implements Mapper<LongWritable, Text, NullWritable, Text> {
 
        @Override
        public void map(LongWritable key, Text value, 
                        OutputCollector<NullWritable, Text> output, 
                        Reporter reporter) throws IOException {
            output.collect(NullWritable.get(), value);
        }
    }
 
    public static class PartitionFormat 
            extends MultipleTextOutputFormat<NullWritable, Text> {
        //和上面一样，就不写了
    }
 
    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        JobConf job = new JobConf(conf, OutputTest.class);
        String[] remainingArgs = 
                new GenericOptionsParser(conf, args).getRemainingArgs();
 
        if (remainingArgs.length != 2) {
            System.err.println("Error!");
            System.exit(1);
        }
 
        Path in = new Path(remainingArgs[0]);
        Path out = new Path(remainingArgs[1]);
 
        FileInputFormat.setInputPaths(job, in);
        FileOutputFormat.setOutputPath(job, out);
 
        job.setJobName("Output");
        job.setMapperClass(MapClass.class);
 
        job.setInputFormat(TextInputFormat.class);
        job.setOutputFormat(PartitionFormat.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
 
        job.setNumReduceTasks(0);
        JobClient.runJob(job);
    }
 
}

将上面的程序打包成jar文件（具体怎么打包，就不说），并在Hadoop2.2.0上面运行（测试数据请在这里下载：http://pan.baidu.com/s/1td8xN）：

/home/q/hadoop-2.2.0/bin/hadoop jar                      \
      /export1/tmp/wyp/OutputText.jar com.wyp.OutputTest \
      /home/wyp/apat63_99.txt                            \ 
      /home/wyp/out

运行完程序之后，可以去/home/wyp/out目录看下运行结果：

[wyp@l-datalog5.data.cn1 ~]$ /home/q/hadoop-2.2.0/bin/hadoop fs         \
                                     -ls /home/wyp/out
.............................这里省略了很多...................................
drwxr-xr-x   - wyp  supergroup     0 2013-11-26 14:25 /home/wyp/out/VE
drwxr-xr-x   - wyp  supergroup     0 2013-11-26 14:25 /home/wyp/out/VG
drwxr-xr-x   - wyp  supergroup     0 2013-11-26 14:25 /home/wyp/out/VN
drwxr-xr-x   - wyp  supergroup     0 2013-11-26 14:25 /home/wyp/out/VU
drwxr-xr-x   - wyp  supergroup     0 2013-11-26 14:25 /home/wyp/out/YE
.............................这里省略了很多...................................
-rw-r--r--   3 wyp  supergroup     0 2013-11-26 14:25 /home/wyp/out/_SUCCESS
 
[wyp@l-datalog5.data.cn1 ~]$ /home/q/hadoop-2.2.0/bin/hadoop fs        \
                                    -ls /home/wyp/out/VN
Found 2 items
-rw-r--r-- 3 wyp supergroup  148 2013-11-26 14:25 /home/wyp/out/VN/part-00000
-rw-r--r-- 3 wyp supergroup  566 2013-11-26 14:25 /home/wyp/out/VN/part-00001
 
[wyp@l-datalog5.data.cn1 ~]$ /home/q/hadoop-2.2.0/bin/hadoop fs        \
                                   -cat /home/wyp/out/VN/part-00001
3430490,1969,3350,1965,"VN","",597185,6,,73,4,43,,0,,,,,,,,,
3630470,1971,4379,1970,"VN","",,1,,244,5,55,,4,,0.375,,22.5,,,,,
3654325,1972,4477,1969,"VN","",,1,,554,1,14,,0,,,,,,,,,
3665081,1972,4526,1970,"VN","",,1,,373,6,66,,1,,0,,3,,,,,
3772710,1973,5072,1972,"VN","",,1,,4,6,65,,1,,0,,8,,,,,
3821853,1974,5296,1971,"VN","",,1,,33,6,69,,1,,0,,23,,,,,
3824277,1974,5310,1970,"VN","",347650,3,,562,1,14,,2,,0.5,,9,,,,0,0
3918104,1975,5793,1972,"VN","",,1,2,4,6,65,5,0,0.4,,0,,18.2,,,,

从上面的结果可以看出，所有country相同的结果都输出到同一个文件夹下面了。MultipleOutputFormat对完全控制文件名和目录名很方便。大家也看到了上面的程序是基于行的split，如果我们要基于列的split，MultipleOutputFormat就无能为力了。这时MultipleOutputs就用上场了。MultipleOutputs在很早的版本就存在，那么我们先看看官方文档是怎么解释MultipleOutputs的：

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.

MultipleOutputs

与MultipleOutputFormat类不一样的是，MultipleOutputs可以为不同的输出产生不同类型，到这里所说的MultipleOutputs类还是旧版本的功能，后面会提到新版本类库的强化版MultipleOutputs类，下面我们来用旧版本的MultipleOutputs类说明它是如何为不同的输出产生不同类型，MultipleOutputs类不是要求给每条记录请求文件名，而是创建多个OutputCollectors。每个OutputCollector可以有自己的OutputFormat和键值对类型，Mapreduce程序将决定如何向每个OutputCollector输出数据（看看上面的英文文档），说的你很晕吧，来看看代码吧！下面的代码将地理相关的信息存储在geo开头的文件中；而将时间相关的信息存储在chrono开头的文件中，具体的代码如下：

package com.wyp;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.MultipleOutputs;
import org.apache.hadoop.util.GenericOptionsParser;
 
import java.io.IOException;
 
/**
 * User: /
 * Date: 13-11-27
 * Time: 下午10:32
 */
public class OldMulOutput {
    public static class MapClass 
            extends MapReduceBase
            implements Mapper<LongWritable, 
            Text, NullWritable, Text> {
        private MultipleOutputs mos;
        private OutputCollector<NullWritable, Text> collector;
 
        public void configure(JobConf conf) {
            mos = new MultipleOutputs(conf);
        }
 
        public void map(LongWritable key, Text value,
                        OutputCollector<NullWritable, Text> output,
                        Reporter reporter) throws IOException {
            String[] arr = value.toString().split(",", -1);
            String chrono = arr[1] + "," + arr[2];
            String geo = arr[4] + "," + arr[5];
            collector = mos.getCollector("chrono", reporter);
            collector.collect(NullWritable.get(), new Text(chrono));
            collector = mos.getCollector("geo", reporter);
            collector.collect(NullWritable.get(), new Text(geo));
        }
 
        public void close() throws IOException {
            mos.close();
        }
 
        public static void main(String[] args) throws IOException {
            Configuration conf = new Configuration();
            String[] remainingArgs =
             new GenericOptionsParser(conf, args).getRemainingArgs();
 
            if (remainingArgs.length != 2) {
                System.err.println("Error!");
                System.exit(1);
            }
             
            JobConf job = new JobConf(conf, OldMulOutput.class);
            Path in = new Path(remainingArgs[0]);
            Path out = new Path(remainingArgs[1]);
            FileInputFormat.setInputPaths(job, in);
            FileOutputFormat.setOutputPath(job, out);
 
            job.setJobName("MultiFile");
            job.setMapperClass(MapClass.class);
            job.setInputFormat(TextInputFormat.class);
            job.setOutputKeyClass(NullWritable.class);
            job.setOutputValueClass(Text.class);
 
            job.setNumReduceTasks(0);
            MultipleOutputs.addNamedOutput(job,
                    "chrono",
                    TextOutputFormat.class,
                    NullWritable.class,
                    Text.class);
 
            MultipleOutputs.addNamedOutput(job,
                    "geo",
                    TextOutputFormat.class,
                    NullWritable.class,
                    Text.class);
            JobClient.runJob(job);
 
        }
    }
}

上面程序来源《Hadoop in action》。同样将上面的程序打包成jar文件（具体怎么打包，也不说了），并在Hadoop2.2.0上面运行（测试数据请在这里下载：http://pan.baidu.com/s/1td8xN）：

/home/q/hadoop-2.2.0/bin/hadoop jar                        \
      /export1/tmp/wyp/OutputText.jar com.wyp.OldMulOutput \
      /home/wyp/apat63_99.txt                              \ 
      /home/wyp/out5

运行完程序之后，可以去/home/wyp/out5目录看下运行结果：

[wyp@l-datalogm1.data.cn1 bin]$ /home/q/hadoop-2.2.0/bin/hadoop fs  \
                                -ls /home/wyp/out5
Found 7 items
-rw-r--r-- 3 wyp sg     0 2013-11-26 14:57 /home/wyp/out5/_SUCCESS
-rw-r--r-- 3 wyp sg 31243 2013-11-26 15:57 /home/wyp/out5/chrono-m-00000
-rw-r--r-- 3 wyp sg 22719 2013-11-26 15:57 /home/wyp/out5/chrono-m-00001
-rw-r--r-- 3 wyp sg 29922 2013-11-26 15:57 /home/wyp/out5/geo-m-00000
-rw-r--r-- 3 wyp sg 20429 2013-11-26 15:57 /home/wyp/out5/geo-m-00001
-rw-r--r-- 3 wyp sg     0 2013-11-26 14:57 /home/wyp/out5/part-m-00000
-rw-r--r-- 3 wyp sg     0 2013-11-26 14:57 /home/wyp/out5/part-m-00001

大家可以看到在输出的文件中还存在以part开头的文件，但是里面没有信息，这是程序默认的输出文件，输出的收集器的名称是不能为part的，这是因为它已经被使用为默认的值了。

从上面的程序可以看出，旧版本的MultipleOutputs可以将文件基于列来进行分割，但是如果你想进行基于行的分割，这就要求你自己去实现代码了，恨不方便，针对这个问题，新版本的MultipleOutputs已经将旧版本的MultipleOutputs和MultipleOutputFormat的功能合并了，也就是说新版本的MultipleOutputs类具有旧版本的MultipleOutputs功能和MultipleOutputFormat功能；同时，在新版本的类库中已经不存在MultipleOutputFormat类了，因为MultipleOutputs都有它的功能了，还要她干嘛呢？看看官方文档是怎么说的：

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.

也就是说MultipleOutputs合并了旧版本的MultipleOutputs功能和MultipleOutputFormat功能，新api都是用mapreduce包。好了，刚刚也说了新版本的MultipleOutputs有了旧版本的MultipleOutputFormat功能，那么我该怎么在新版的MultipleOutputs中实现旧版本MultipleOutputFormat的多文件输出呢？也就是上面第一个程序。看看下面的代码吧。

package com.wyp;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
 
import java.io.IOException;
 

public class MulOutput {
    public static class MapClass 
            extends Mapper<LongWritable, Text, NullWritable, Text> {
        private MultipleOutputs mos;
        @Override
        protected void setup(Context context) 
                throws IOException, InterruptedException {
            super.setup(context);
            mos = new MultipleOutputs(context);
        }
 
        @Override
        protected void map(LongWritable key, 
                           Text value, 
                           Context context) 
                throws IOException, InterruptedException {
            mos.write(NullWritable.get(), value, 
                    generateFileName(value));
        }
 
        private String generateFileName(Text value) {
            String[] split = value.toString().split(",", -1);
            String country = split[4].substring(1, 3);
            return country + "/";
        }
 
        @Override
        protected void cleanup(Context context) 
                throws IOException, InterruptedException {
            super.cleanup(context);
            mos.close();
        }
    }
 
    public static void main(String[] args) 
            throws IOException, ClassNotFoundException, 
            InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "MulOutput");
        String[] remainingArgs = 
                new GenericOptionsParser(conf, args)
                        .getRemainingArgs();
 
        if (remainingArgs.length != 2) {
            System.err.println("Error!");
            System.exit(1);
        }
        Path in = new Path(remainingArgs[0]);
        Path out = new Path(remainingArgs[1]);
 
        FileInputFormat.setInputPaths(job, in);
        FileOutputFormat.setOutputPath(job, out);
 
        job.setJarByClass(MulOutput.class);
        job.setMapperClass(MapClass.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        job.setNumReduceTasks(0);
 
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

上面的程序通过setup(Context context)来初始化MultipleOutputs对象，并在mapper函数中调用MultipleOutputs的write方法将数据输出到根据value的值不同的文件夹中（通过调用generateFileName函数来处理）。MultipleOutputs类有多个不同版本的write方法，它们的函数原型如下：

public <K, V> void write(String namedOutput, K key, V value)
      throws IOException, InterruptedException
 
public <K, V> void write(String namedOutput, K key, V value,
      String baseOutputPath) throws IOException, InterruptedException 
 
public void write(KEYOUT key, VALUEOUT value, String baseOutputPath) 
      throws IOException, InterruptedException

我们可以根据不同的需求调用不同的write方法。
好了，大家来看看上面程序运行的结果吧：

/home/q/hadoop-2.2.0/bin/hadoop jar                        \
      /export1/tmp/wyp/OutputText.jar com.wyp.MulOutput    \
      /home/wyp/apat63_99.txt                              \ 
      /home/wyp/out11

测试数据还是上面给的地址。看下/home/wyp/out11文件中有什么吧：

[wyp@l-datalogm1.data.cn1 bin]$ /home/q/hadoop-2.2.0/bin/hadoop fs \
                                            -ls /home/wyp/out11
.............................这里省略了很多...................................
drwxr-xr-x - wyp  supergroup  0 2013-11-26 19:42 /home/wyp/out11/VN
drwxr-xr-x - wyp  supergroup  0 2013-11-26 19:41 /home/wyp/out11/VU
drwxr-xr-x - wyp  supergroup  0 2013-11-26 19:42 /home/wyp/out11/YE
drwxr-xr-x - wyp  supergroup  0 2013-11-26 19:42 /home/wyp/out11/YU
drwxr-xr-x - wyp  supergroup  0 2013-11-26 19:42 /home/wyp/out11/ZA
.............................这里省略了很多...................................
-rw-r--r-- 3 wyp  supergroup  0 2013-11-26 19:42 /home/wyp/out11/_SUCCESS
-rw-r--r-- 3 wyp  supergroup  0 2013-11-26 19:42 /home/wyp/out11/part-m-00000
-rw-r--r-- 3 wyp  supergroup  0 2013-11-26 19:42 /home/wyp/out11/part-m-00001

　　现在输出的结果和用旧版本的MultipleOutputFormat输出的结果很类似了；但是在输出的结果中还有两个以part开头的文件夹，而且里面什么都没有，这是怎么回事？和第二个测试程序一样，这也是程序默认的输出文件名。那么我们可以在程序输出的结果中不输出两个文件夹吗？当然可以了，呵呵。如何实现呢？其实很简单，在上面的代码的main函数中加入以下一行代码：

LazyOutputFormat.setOutputFormatClass(job,
                TextOutputFormat.class);

如果加入了上面的一行代码，请同时注释掉你代码中下面一行代码（如果有）

job.setOutputFormatClass(TextOutputFormat.class);

再去看下输出结果吧：

[wyp@l-datalogm1.data.cn1 bin]$ /home/q/hadoop-2.2.0/bin/hadoop fs \
                                                  -ls /home/wyp/out12
.............................这里省略了很多...................................
drwxr-xr-x - wyp supergroup    0 2013-11-26 19:44 /home/wyp/out12/VU
drwxr-xr-x - wyp supergroup    0 2013-11-26 19:44 /home/wyp/out12/YE
drwxr-xr-x - wyp supergroup    0 2013-11-26 19:44 /home/wyp/out12/YU
drwxr-xr-x - wyp supergroup    0 2013-11-26 19:44 /home/wyp/out12/ZA
drwxr-xr-x - wyp supergroup    0 2013-11-26 19:44 /home/wyp/out12/ZM
drwxr-xr-x - wyp supergroup    0 2013-11-26 19:44 /home/wyp/out12/ZW
.............................这里省略了很多...................................
-rw-r--r-- 3 wyp supergroup    0 2013-11-26 19:44 /home/wyp/out12/_SUCCESS
Author: Tunan
Link: http://yerias.github.io/2020/11/25/hadoop/15/
Copyright Notice: All articles in this blog are licensed under CC BY-NC-SA 4.0 unless stating additionally.
Hadoop