第一个参数: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