ES实现百亿级数据实时分析实战案例

本文主要是介绍ES实现百亿级数据实时分析实战案例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

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

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

背景

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

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

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

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

技术方案

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

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

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

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

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

技术方案由此确定。

技术架构

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

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

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

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

代码实现

异步任务

// 启动并行任务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()){// 省略结果处理流程}

指标计算

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