第一个参数:initScans(job)

private List<Scan> initScans(Job job) {
        // 时间戳+....
        Configuration conf = job.getConfiguration();
        // 获取运行时间: yyyy-MM-dd
        String date = conf.get(GlobalConstants.RUNNING_DATE_PARAMES);
        long startDate = TimeUtil.parseString2Long(date);
        long endDate = startDate + GlobalConstants.DAY_OF_MILLISECONDS;

        Scan scan = new Scan();
        // 定义hbase扫描的开始rowkey和结束rowkey
        scan.setStartRow(Bytes.toBytes("" + startDate));
        scan.setStopRow(Bytes.toBytes("" + endDate));

        FilterList filterList = new FilterList();
        // 过滤数据，只分析launch事件
        filterList.addFilter(new SingleColumnValueFilter(Bytes
                .toBytes(EventLogConstants.EVENT_LOGS_FAMILY_NAME), Bytes
                .toBytes(EventLogConstants.LOG_COLUMN_NAME_EVENT_NAME),
                CompareOp.EQUAL, Bytes.toBytes(EventEnum.LAUNCH.alias)));
        // 定义mapper中需要获取的列名
        String[] columns = new String[] {
                EventLogConstants.LOG_COLUMN_NAME_EVENT_NAME,
                EventLogConstants.LOG_COLUMN_NAME_UUID,
                EventLogConstants.LOG_COLUMN_NAME_SERVER_TIME,
                EventLogConstants.LOG_COLUMN_NAME_PLATFORM,
                EventLogConstants.LOG_COLUMN_NAME_BROWSER_NAME,
                EventLogConstants.LOG_COLUMN_NAME_BROWSER_VERSION };

        // scan.addColumn(family, qualifier)
        filterList.addFilter(this.getColumnFilter(columns));

        scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME,
                Bytes.toBytes(EventLogConstants.HBASE_NAME_EVENT_LOGS));

        scan.setFilter(filterList);

        return Lists.newArrayList(scan);
    }

获取配置

Configuration conf = job.getConfiguration();

// 获取运行时间: yyyy-MM-dd processArgs方法里面的conf.set

String date = conf.get(GlobalConstants.RUNNING_DATE_PARAMES);

将字符转换成long类型的时间 TimeUtil类完整方法

long startDate = TimeUtil.parseString2Long(date);
long endDate = startDate + GlobalConstants.DAY_OF_MILLISECONDS;

DAY_OF_MILLISECONDS 一天有多少个毫秒

Scan scan = new Scan(); // 定义hbase扫描的开始rowkey和结束rowkey scan.setStartRow(Bytes.toBytes("" + startDate)); scan.setStopRow(Bytes.toBytes("" + endDate));

FilterList filterList = new FilterList();

// 过滤数据，只分析launch事件
FilterList 类的实现

final public class FilterList extends Filter public abstract class Filter

public void addFilter(Filter filter)

    // 过滤数据，只分析launch事件
    filterList.addFilter(new SingleColumnValueFilter(Bytes
            .toBytes(EventLogConstants.EVENT_LOGS_FAMILY_NAME), Bytes
            .toBytes(EventLogConstants.LOG_COLUMN_NAME_EVENT_NAME),
            CompareOp.EQUAL, Bytes.toBytes(EventEnum.LAUNCH.alias)));

filterList.addFilter(new SingleColumnValueFilter(

/**

Constructor for binary compare of the value of a single column. If the
column is found and the condition passes, all columns of the row will be
emitted. If the condition fails, the row will not be emitted.
<p>
Use the filterIfColumnMissing flag to set whether the rest of the columns
in a row will be emitted if the specified column to check is not found in
the row.
@param family name of column family
@param qualifier name of column qualifier
@param compareOp operator
@param value value to compare column values against */ public SingleColumnValueFilter(final byte [] family, final byte [] qualifier, final CompareOp compareOp, final byte[] value) { this(family, qualifier, compareOp, new BinaryComparator(value)); } }

Bytes.toBytes(EventLogConstants.EVENT_LOGS_FAMILY_NAME),
EventLogConstants

/**
 * event_logs表的列簇名称
 */
public static final String EVENT_LOGS_FAMILY_NAME = "log";

Bytes.toBytes(EventLogConstants.LOG_COLUMN_NAME_EVENT_NAME),

/** * 事件名称 */ public static final String LOG_COLUMN_NAME_EVENT_NAME = "en";

CompareOp.EQUAL, 
是在CompareFilter 中定义的

public abstract class Filter { public abstract class FilterBase extends Filter { public abstract class CompareFilter extends FilterBase {

public abstract class CompareFilter extends FilterBase {

/** Comparison operators. / @InterfaceAudience.Public @InterfaceStability.Stable public enum CompareOp { /* less than / LESS, /* less than or equal to / LESS_OR_EQUAL, /* equals / EQUAL, /* not equal / NOT_EQUAL, /* greater than or equal to / GREATER_OR_EQUAL, /* greater than / GREATER, /* no operation */ NO_OP, }

Bytes.toBytes(EventEnum.LAUNCH

/** * 事件枚举类。指定事件的名称 * * @author root * */ public static enum EventEnum { LAUNCH(1, "launch event", "e_l"), // launch事件，表示第一次访问 PAGEVIEW(2, "page view event", "e_pv"), // 页面浏览事件 CHARGEREQUEST(3, "charge request event", "e_crt"), // 订单生产事件 CHARGESUCCESS(4, "charge success event", "e_cs"), // 订单成功支付事件 CHARGEREFUND(5, "charge refund event", "e_cr"), // 订单退款事件 EVENT(6, "event duration event", "e_e") // 事件 ;

    public final int id; // id 唯一标识
    public final String name; // 名称
    public final String alias; // 别名，用于数据收集的简写

// 定义mapper中需要获取的列名 String[] columns = new String[] { EventLogConstants.LOG_COLUMN_NAME_EVENT_NAME, //事件名称en EventLogConstants.LOG_COLUMN_NAME_UUID, //用户UUID u_ud EventLogConstants.LOG_COLUMN_NAME_SERVER_TIME, //服务器时间s_time EventLogConstants.LOG_COLUMN_NAME_PLATFORM, //用户平台pl EventLogConstants.LOG_COLUMN_NAME_BROWSER_NAME, //browser EventLogConstants.LOG_COLUMN_NAME_BROWSER_VERSION //browser_v };

filterList.addFilter(this.getColumnFilter(columns));

/** * 获取这个列名过滤的column * * @param columns * @return */ private Filter getColumnFilter(String[] columns) { int length = columns.length; byte[][] filter = new byte[length][]; for (int i = 0; i < length; i++) { filter[i] = Bytes.toBytes(columns[i]); } return new MultipleColumnPrefixFilter(filter); }

>scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME,

// If an application wants to use multiple scans over different tables each scan must // define this attribute with the appropriate table name by calling // scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(tableName)) static public final String SCAN_ATTRIBUTES_TABLE_NAME = "scan.attributes.table.name";

>Bytes.toBytes(EventLogConstants.HBASE_NAME_EVENT_LOGS));

/** * 表名称 */ public static final String HBASE_NAME_EVENT_LOGS = "eventlog";

    scan.setFilter(filterList);

    return Lists.newArrayList(scan);
}

项目入口
http://www.jianshu.com/p/aa3b12b0d426