ES实现百亿级数据实时分析实战案例-技术圈

点击上方蓝色字体，选择“设为星标”

回复”资源“获取更多资源

背景

我们小组前段时间接到一个需求，希望能够按照小时为单位，看到每个实验中各种特征（单个或组合）的覆盖率、正样本占比、负样本占比。我简单解释一下这三种指标的定义：

覆盖率：所有样本中出现某一特征的样本的比例
正样本占比：所有出现该特征的样本中，正样本的比例
负样本占比：所有出现该特征的样本中，负样本的比例

光看这三个指标，大家可能会觉得这个需求很简单，无非就是一个简单的筛选、聚合而已。

如果真的这么简单，我也没必要写这篇文章单独记录了。问题的关键就在于，每小时有将近1亿的数据量，而我们需要保存7天的数据，数据总量预计超过了100亿。

技术方案

在了解清楚需求后，我们小组马上对技术方案展开讨论，讨论过程中出现了3种方案：

第一种：用Spark流式计算，计算每一种可能单个或组合特征的相关指标
第二种：收到客户端请求后，遍历HDFS中相关数据，进行离线计算
第三种：将数据按照实验+小时分索引存入ES，收到客户端请求后，实时计算返回

首先，第一种方案直接被diss，原因是一个实验一般会出现几百、上千个特征，而这些特征的组合何止几亿种，全部计算的话，可行性暂且不论，光是对资源的消耗就无法承受。

第二种方案，虽然技术上是可行的，但离线计算所需时间较长，对用户来说，体验并不理想。并且，为了计算目标1%的数据而要遍历所有数据，对资源也存在很大浪费。

第三种方案，将数据按照实验+小时分索引后，可以将每个索引包含的数据量降到1000万以下，再借助ES在查询、聚合方面高效的能力，应该可以实现秒级响应，并且用户体验也会非常好。

技术方案由此确定。

技术架构

1.用Spark从Kafka中接入原始数据，之后对数据进行解析，转换成我们的目标格式

2.将数据按照实验+小时分索引存入ES中

3.接受到用户请求后，将请求按照实验+特征+小时组合，创建多个异步任务，由这些异步任务并行从ES中过滤并聚合相关数据，得到结果

4.将异步任务的结果进行合并，返回给前端进行展示

代码实现

异步任务

// 启动并行任务

final Map>> 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> 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 future = TaskExecutor.submit(task);

 resultList.add(future);

 }

futures.put(name, resultList);

 }

}

}

final QueryRes queryRes = new QueryRes();

final Iterator>>> it = futures.entrySet().iterator();

while (it.hasNext()){

// 省略结果处理流程

 }

指标计算

// 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;

聚合

int label = 0;

final ExistsQueryBuilder existsQuery = QueryBuilders.existsQuery(fieldName);

// 包含指定特征的正样本数量

final BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

final List 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中部分索引的信息：

突破性能瓶颈！ElasticSearch百亿级数据检索优化案例

ElasticSearch读写底层原理及性能调优

一文俯瞰Elasticsearch核心原理

文章不错？点个【在看】吧！ 👇