PathFilter.java

call.java

//在main job里进行如下调用来激活path filter
FileInputFormat.setInputPathFilter(job, TongSeqFilter.class);

PathFilter.java

package com.miaozhen.dataeye.input;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
/**
 * 过滤掉文件名不包含 conf.get("org.stackbox.filePattern", "campaign_")的日志文件
 */
public class TongSeqFilter extends Configured implements PathFilter {

    String filePattern = null;
    Configuration conf;
    FileSystem fs;

    public boolean accept(Path path) {
        try
        {
            if(!fs.isFile(path))
            {
                return true;
            }
            else
            {
                return path.toString().contains(filePattern);
            }
        }
        catch(IOException ex)
        {
            ex.printStackTrace();
            return false;
        }
    }


    @Override
    public void setConf(Configuration conf) {
        this.conf = conf;
        if(conf != null)
        {
            try
            {
                fs = FileSystem.get(conf);
                filePattern = conf.get("org.stackbox.filePattern", "campaign_");
            }
            catch(IOException ex)
            {
                ex.printStackTrace();
            }
        }
        System.out.println("Pattern is  " + filePattern);

        System.out.println("===========>");
    }
}