hadoop - 可以使 Hadoop MultipleInputs.addInputPath 递归工作吗？

Question

最近版本的 Hadoop 已经使用 FileInputFormat.setInputDirRecursive 轻松支持嵌套输入目录，它依赖于 mapreduce.input.fileinputformat.input.dir.recursive 配置键。

也可以使用 MultipleInputs.addInputPath 指定多个映射器/输入目录组合。

但是我可以同时做这两件事吗？换句话说，有没有一种方法可以指定多个映射器/输入目录组合，其中递归地包含输入目录？

一个具体的例子:我有以下目录结构:

• /dataset1/subdir1/data1.txt
• /dataset2/subdir2/data2.txt

我试过这样的:

Job job = Job.getInstance(conf);
FileInputFormat.setInputDirRecursive(job, true);
MultipleInputs.addInputPath(job, new Path("/dataset1"), TextInputFormat.class,
    Mapper1.class);
MultipleInputs.addInputPath(job, new Path("/dataset2"), TextInputFormat.class,
    Mapper2.class);
...
job.waitForCompletion(true);

但随后我得到了一个异常 Error: java.io.IOException: 's3://bucketname/dataset1/subdir1' is a directory

这是在 Hadoop 2.4.0 下的 Amazon EMR 中运行的。

编辑:Hadoop 版本是 2.4.0，不是 2.6.0

Answer 1

嗯，不确定 s3，但这很正常。应指向文件 而不是目录。

试试这个。

方法一

final static public void addInputPathRecursively(FileSystem fs, Path path, PathFilter inputFilter, Job job,boolean swithc) throws IOException
        {
            for (FileStatus stat : fs.listStatus(path, inputFilter))
            {
                if (stat.isDirectory())
                {

                    addInputPathRecursively(fs, stat.getPath(), inputFilter, job);
                } else
                {
                    if (swithc)
                    {
                        MultipleInputs.addInputPath(job, new Path(stat.getPath().toString()), TextInputFormat.class, Mapper1.class);

                    } else
                        MultipleInputs.addInputPath(job, new Path(stat.getPath().toString()), TextInputFormat.class, Mapper2.class);
                }
            }
        }

在驱动类中你可以相应地调用它。

addInputPathRecursively(fs, datset1path, new FileFilter(conf, fs,
                        new String[] { txt }), job,true);

addInputPathRecursively(fs, datset2path, new FileFilter(conf, fs,
                        new String[] { txt }), job,false);

这是一个示例，但如果您想应用正则表达式，则可以正确控制路径过滤器。

• 方法二设置它也应该有魔力。

  FileInputFormat.setInputDirRecursive(job, true);

• 方法三绕过映射器内部并在行级处理。 (这不是个好主意!)