在hadoop中限制没有reducer的MultipleOutput中的mapper数量

Question

您好，我有一个从 HBase 读取记录并写入文本文件的应用程序 HBase 表有 200 个区域。 我在映射器类中使用 MultipleOutputs 写入多个文件，并且我正在从传入记录中创建文件名。

我正在制作 40 个唯一的文件名。 我能够正确获取记录，但我的问题是，当 mapreduce 完成时，它会创建 40 个文件以及 2k 个具有正确名称但附加的额外文件 与 m-000 等等。

这是因为我有 200 个区域，MultipleOutputs 为每个映射器创建文件，因此 200 个映射器，每个映射器有 40 个唯一文件，这就是它创建 40*200 个文件的原因。

如果没有自定义分区器，我不知道如何避免这种情况。

有没有办法强制将记录写入所属文件而不是拆分为多个文件。

我使用了自定义分区器类，它工作正常，但我不想使用它，因为我只是从 HBase 读取而不进行减速器操作。此外，如果我必须创建任何额外的文件名，那么我必须更改我的代码也是。

这是我的映射器代码

   public class DefaultMapper extends TableMapper<NullWritable, Text> {
        private Text text = new Text();
        MultipleOutputs<NullWritable, Text> multipleOutputs;
        String strName = "";

        @Override()
        public void setup(Context context) throws java.io.IOException, java.lang.InterruptedException {
            multipleOutputs = new MultipleOutputs<NullWritable, Text>(context);
        }
String FILE_NAME = new String(value.getValue(Bytes.toBytes(HbaseBulkLoadMapperConstants.COLUMN_FAMILY),Bytes.toBytes(HbaseBulkLoadMapperConstants.FILE_NAME)));

        multipleOutputs.write(NullWritable.get(), new Text(text.toString()),FILE_NAME);
        //context.write(NullWritable.get(), text);
    }

没有减速器类

这就是我的输出的样子，理想情况下应该只创建一个Japan.BUS.gz 文件。其他文件也是非常小的文件

Japan.BUS-m-00193.gz
Japan.BUS-m-00194.gz
Japan.BUS-m-00195.gz
Japan.BUS-m-00196.gz

Answer 1

我也遇到过同样的情况，也解决了。

MultipleOutputs multipleOutputs = null;

String keyToFind = new String();

    public void setup(Context context) throws IOException, InterruptedException
    {
        this.multipleOutputs_normal = new MultipleOutputs<KEYOUT, VALUEOUT>(context);
    }

public void map(NullWritable key , Text values, Context context) throws IOException, InterruptedException
{

   String valToFindInCol[] = values.toString.split(",");/** Lets say comma seperated **/

    if (keyToFind .equals(valToFindInCol[2].toString())|| keyToFind == null) /** Say you need to match 2 position element **/
    {
        this.multipleOutputs.write(NullWritable.get(),<valToWrite>, valToFindInCol[2]);
    } 
    else 
    {
        this.multipleOutputs.close();
        this.multipleOutputs = null;
        this.multipleOutputs = new MultipleOutputs<KEYOUT, VALUEOUT>(context);

        this.multipleOutputs.write(NullWritable.get(),<valToWrite>, valToFindInCol[2]);

    }

    keyToFind=valToFindInCol[2];
}