背景
我们小组前段时间接到一个需求，希望能够按照小时为单位，看到每个实验中各种特征（单个或组合）的覆盖率、正样本占比、负样本占比。我简单解释一下这三种指标的定义：
- 覆盖率：所有样本中出现某一特征的样本的比例
- 正样本占比：所有出现该特征的样本中，正样本的比例
- 负样本占比：所有出现该特征的样本中，负样本的比例
光看这三个指标，大家可能会觉得这个需求很简单，无非就是一个简单的筛选、聚合而已。
如果真的这么简单，我也没必要写这篇文章单独记录了。问题的关键就在于，每小时有将近1亿的数据量，而我们需要保存7天的数据，数据总量预计超过了100亿。
技术方案
在了解清楚需求后，我们小组马上对技术方案展开讨论，讨论过程中出现了3种方案：
- 第一种：用Spark流式计算，计算每一种可能单个或组合特征的相关指标
- 第二种：收到客户端请求后，遍历HDFS中相关数据，进行离线计算
- 第三种：将数据按照实验 小时分索引存入ES，收到客户端请求后，实时计算返回
首先，第一种方案直接被diss，原因是一个实验一般会出现几百、上千个特征，而这些特征的组合何止几亿种，全部计算的话，可行性暂且不论，光是对资源的消耗就无法承受。
第二种方案，虽然技术上是可行的，但离线计算所需时间较长，对用户来说，体验并不理想。并且，为了计算目标1%的数据而要遍历所有数据，对资源也存在很大浪费。
第三种方案，将数据按照实验 小时分索引后，可以将每个索引包含的数据量降到1000万以下，再借助ES在查询、聚合方面高效的能力，应该可以实现秒级响应，并且用户体验也会非常好。
技术方案由此确定。
技术架构
1.用Spark从Kafka中接入原始数据，之后对数据进行解析，转换成我们的目标格式
2.将数据按照实验 小时分索引存入ES中
3.接受到用户请求后，将请求按照实验 特征 小时组合，创建多个异步任务，由这些异步任务并行从ES中过滤并聚合相关数据，得到结果
4.将异步任务的结果进行合并，返回给前端进行展示
代码实现
异步任务
代码语言：javascript复制// 启动并行任务
final Map<String,List<Future<GetCoverageTask.Result>>> futures = Maps.newHashMap();
for(String metric : metrics) { // 遍历要计算的指标
final SampleRatio sampleRatio = getSampleRatio(metric);
for (String exptId : expts) { // 遍历目标实验列表
for (String id : features) { // 遍历要分析的特征
final String name = getMetricsName(exptId, sampleRatio, id);
final List<Future<GetCoverageTask.Result>> resultList = Lists.newArrayList();
for (Date hour : coveredHours) { // 将时间按照小时进行拆分
final String fieldName = getFieldName(isFect ? Constants.FACET_COLLECT : Constants.FEATURE_COLLECT, id);
final GetCoverageTask task = new GetCoverageTask(exptId, fieldName, sampleRatio, hour);
// 启动并行任务
final Future<GetCoverageTask.Result> future = TaskExecutor.submit(task);
resultList.add(future);
}
futures.put(name, resultList);
}
}
}
final QueryRes queryRes = new QueryRes();
final Iterator<Map.Entry<String, List<Future<GetCoverageTask.Result>>>> it = futures.entrySet().iterator();
while (it.hasNext()){
// 省略结果处理流程
}指标计算
代码语言：javascript复制// 1. 对文档进行聚合运行，分别得到基础文档的数量，以及目标文档数量
final AggregationBuilder[] agg = getAggregationBuilder(sampleRatio, fieldName);
final SearchSourceBuilder searchBuilder = new SearchSourceBuilder();
searchBuilder.aggregation(agg[0]).aggregation(agg[1]).size(0);
// 2. 得到覆盖率
final String indexName = getIndexName(exptId, hour);
final Search search = new Search.Builder(searchBuilder.toString())
.addIndex(indexName).addType(getType()).build();
final SearchResult result = jestClient.execute(search);
if(result.getResponseCode() != HttpUtils.STATUS_CODE_200){
// 请求出错
log.warn(result.getErrorMessage());
return 0f;
}
final MetricAggregation aggregations = result.getAggregations();
// 3. 解析结果
final long dividend ;
if(SampleRatio.ALL == sampleRatio){
dividend = aggregations.getValueCountAggregation(Constants.DIVIDEND).getValueCount();
}else {
dividend = aggregations.getFilterAggregation(Constants.DIVIDEND).getCount();
}
// 防止出现被除数为0时程序异常
if(dividend <= 0){
return 0f;
}
long divisor = aggregations.getFilterAggregation(Constants.DIVISOR).getCount();
return divisor / (float)dividend;聚合
代码语言：javascript复制int label = 0;
final ExistsQueryBuilder existsQuery = QueryBuilders.existsQuery(fieldName);
// 包含指定特征的正样本数量
final BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
final List<QueryBuilder> must = boolQuery.must();
// 计算样本数量
TermQueryBuilder labelQuery = null;
if(SampleRatio.POSITIVE == sampleRatio) {
// 计算正样本数量
label = 1;
labelQuery = QueryBuilders.termQuery(Constants.LABEL, label);
must.add(labelQuery);
}else if(SampleRatio.NEGATIVE == sampleRatio) {
// 计算负样本数量
labelQuery = QueryBuilders.termQuery(Constants.LABEL, label);
must.add(labelQuery);
}
must.add(existsQuery);
final ValueCountAggregationBuilder existsCountAgg = AggregationBuilders.count(sampleRatio.getField());
existsCountAgg.field(fieldName);
final FilterAggregationBuilder filterAgg = AggregationBuilders.filter(aggName, boolQuery);
filterAgg.subAggregation(existsCountAgg);
return filterAgg;上线效果
上线后表现完全满足预期，平均请求耗时在3秒左右，用户体验良好。感谢各位小伙伴的辛苦付出~~
下图是ES中部分索引的信息：
