Hadoop 之 PathFilter 的一些坑

最近再写一个MapReduce的时候出现了一个诡异的问题, PathFilter无法生效, 具体描述如下:

代码参考的是另外一个项目(用的是SequenceFileInputFormat),其PathFilter能正常工作
我写的这个MR用的公司的一个CombineFileInputFormat, 虽然设置了Filter但是程序运行的时候PathFilter甚至都没实例化。

感觉从PathFilter初始化这个点找应该是个正确的方向,As we all know, FileInputFormat 是所有XXFileInputFormat的父类,果然在其中找到了如下代码

public abstract class FileInputFormat<K, V> extends InputFormat<K, V>  {

    public List<InputSplit> getSplits(JobContext job) throws IOException {
            //other code
            List<FileStatus> files = listStatus(job);
            //other code
    }

    protected List<FileStatus> listStatus(JobContext job) throws IOException {
            //other code
            Path[] dirs = getInputPaths(job);
            //other code
    }

    public static PathFilter getInputPathFilter(JobContext context) {
       Configuration conf = context.getConfiguration();
       Class<?> filterClass = conf.getClass(PATHFILTER_CLASS, null,
           PathFilter.class);
       return (filterClass != null) ?
           (PathFilter) ReflectionUtils.newInstance(filterClass, conf) : null;
     }
 }

这样的话, 只要确保 getSplits 方法调用了 getInputPathFilter , 那么PathFilter便能初始化成功. 所以实现自定义FileInputFormat

的时候要注意override 方法的实现,先看看CombineFileInputFormat和SequenceFileInputFormat是怎么实现的吧

CombineFileInputFormat: 重写了getSplits方法, 但是在重写的方法里调用了 super.listStatus(job), 所以PathFilter正常
SequenceFileInputFormat: 重写了listStatus方法, 但是在重写方法里调用了 super.listStatus(job), 所以PathFilter也依然正常
那么再看看自己写的那个InputFormat, 也重写了getSplits方法, 但是获取block status是用 getInputPaths获取路径然后手动获取status的, 根本就没管父类的listSt方法，因此导致PathFilter失效

这个是一个PathFiter的 Demo

THE END

声明: 本文 “PathFilter无法生效” 采用 CC BY-NC-SA 4.0 协议进行授权.

转载请注明原文地址: http://stackbox.org/2015-07-hdfs-pathfilter-not-work/index.html