Elasticsearch在物流数据中心的应用

Elasticsearch ，简称es，主要运用于全文搜索、数据分析， 底层使用开源库Lucene，拥有丰富的REST API，开箱即用。分布式的数据存储、倒排索引等设计，使其可以快速存储、搜索、分析海量数据。典型的使用方和应用场景，如github,StackOverflow，elasticsearch+logstash+kibana 一体化的日志分析。

下面主要从我们如何存储数据、用这些数据干什么两个方面来展开我们对于es的应用。有啥子不合理、不足之处欢迎大家指正。

背景

随着饿了么运单数据的增长，传统的数据库很难支撑现有的业务：

1、各种场景的数据查询、统计，导致数据库必须加入各种字段的索引，大大增加了在生成一条运单，插入数据所需要的成本，严重影响了生成运单的并发度；大量的索引同时占用了很大的磁盘空间，同时给数据库变更带来的更大的风险；

2、很多场景没有办法或者很难实现，如：运单分页查询。大体量的数据数据库必然是sharding的，此时数据在分页上面必须需要通过别的工具。此时我们引入了ES，来处理允许一定延迟的数据查询、统计的业务。

数据存储

如大家所知的那样，ES不支持事务、复杂的数据关系（后期版本稍有改善，但是仍然支持的不是很好），利用_version (版本号)的方式来确保应用中相互冲突的变更不会导致数据丢失，那么我们是如何存储我们的数据，数据结构是什么样子，如何保证数据的完整性和一致性的呢？

一、数据结构

首先说下我们的运单数据索引的数据结构。

1、合适的分片数和副本数。网上有很多关于如何规划分片数的文章，本人感觉可以作为参考， 在机器性能、数据量的大小、使用场景等等的不同，分片、副本的数量最好可以通过压测或者是线上实际流量来做调整。

2、我们会尽量减少我们所需要的字段，做到够用就好，mapping设置方面：设置"_all"为false，String类型"index"尽量设置为不分词（"not_analyzed"，根据需要设置analyzed），商家名称这类String类型字段只存储索引结构，不存储原始文档（后面会聊到如何拿到原始文档）。

起初我们在建运单索引的时候，我们是尽量冗余运单上面的所有信息，导致一个星期的运单数据达到一个T的大小，而上面大部分的字段都是不需要的，磁盘利用率很低，而用于该集群的都是ssd盘，常常由于磁盘存不下，而需要添加机器，导致大量的资源浪费。这也需要我们支持一个额外的能力，万一需要添加某个运单字段，我们需要在需求上线之前迅速将历史数据补齐这个字段，同时不影响线上。（我们现在可以一个晚上重刷我们需要周期内的历史数据）。

"mappings": {
	"index_type_name": {
    "_all": {
      "enabled": false
    },
    "_source": {
      "excludes": ["merchant_name"]
    },
    "properties": {
      "order_id": {
        "type": "long"
      },
	  "merchant_name": {
        "type": "string",
        "index": "not_analyzed"
      }
     ...
    }
  }
}
复制代码

3、以一天为一个索引（根据业务场景，因为我们的业务场景大部分要的都是某天的数据），这也为我们根据实际线上流量调整我们分片、副本数量提供了方便，修改完索引的模板（"_template"）之后，第二天会自动生效，而查询多天不同分片数量的运单索引的联合查询不会影响查询结果。

4、尽量避免Nested Objects数据类型(nested数据结构)。每一个nested object 将会作为一个隐藏的单独文本建立索引，虽然官网上说在查询的时候将根文本和nested文档拼接是很快的，就跟把他们当成一个单独的文本一样的快。但是其实还是有一部分的额外的消耗，尤其是在aggs聚合的时候，它会使一层聚合其实变成了两层聚合：需要先聚合隐藏文件，再对实际需求进行聚合。如果真的需要放入数组类型的数据，可以根据实际需求，转化为一个字段，直接建在主数据上面（有必要的话，可以对nested object直接建一个新的索引）。

比如：我们现在有一个索引，里面有某个学校每天每个学生的学习、生活情况，每个学生每天会产生一条数据。现在我们想统计每个班级某天 中午在校吃饭的人数、以及一天在校的用餐次数，我们可以设计一个nested Objects数据结构来存储一天三餐的情况，也可以在主数据上添加四个字段：早上是否在校吃饭，中午是否在校吃饭、晚上是否在校吃饭，三餐在校用餐次数，这样就可以直接对着这四个字段进行数据统计。

5、尽量减少script line的使用。同样的道理，我们可以预先将需要用script line 的中间值先存到主数据上面。避免查询、统计时候的额外消耗。

二、数据如何存储的

1、考虑在不影响已有的业务情况下，我们采取解析运单数据落库产生的binlog日志来建索引（binlog日志公司有一套解决方案，不一定非要使用binlog日志，运单状态变化的mq消息也是可以的）， 使其与运单正常业务解耦 。

2、此时我们不会直接拿这条数据插入ES，因为运单状态变化在同一个时刻可能会发生多次，每次的数据插入不一定是数据库当前的状态，而且不论binlog日志、还是运单状态变化消息都只是涵盖了部分数据，如果要运单在发消息的时候，把所有需要的数据补齐，对于运单的业务来说，会面临经常修改消息结构的问题，这已经违背了我们要使其与运单正常业务解耦的初衷， 所以我们在收到这条数据变更的时候， 会通过运单id反查运单数据 ，运单肯定会时时刻刻保持有最新的通过运单id查询运单全部信息的接口，这样我们就可以拿到我们想要的任何最新数据。

3、通过es建索引的 bulk api 减少与es集群的交互次数，提高数据写入的吞吐量。

4、同一条运单数据，在同一个时刻可能会在机器A和机器B中同时发生更新操作，机器A查询到的是旧数据，机器B查询到了新数据，但是写入索引的时候机器B先写入ES集群，机器A后写入集群，导致数据错误。解决方案： 每条数据写入的时候，添加一个分布式锁 ，相同运单号的数据在同一个时刻只能有一条发生写索引的动作，没有获得分布式锁消息，丢入延迟队列，下次再消费。

5、数据的补偿（此处就不展开了）。

数据的查询和统计

一、运单数据查询

前面我们讲述了我们ES中的索引结构遵循的一些原则，其中有一条是，我们不会在ES中存储原始文档，那么我们是如何支持查询运单的具体数据的呢？其实这就是一个ES集群的定位问题，我们的ES集群仅仅是用来丰富运单查询、支持数据统计的功能，我们并不支持数据的实际存储，我们存储的仅仅只是每个字段的索引而已，通过每个字段的索引支持各种各样的运单查询、数据统计，如果需要查询运单的详细信息，我们通过ES查询得到运单id后，会去运单的查询服务查询到该信息，再吐给需求方，我们会将这个步骤包掉，需求方无感知，且返回的数据只是将运单的查询服务的数据包了一层，尽可能减少其他方的接入成本。

二、基于ES的数据的统计

ES在做数据统计的时候往往会很消耗ES集群的资源，所以我们通常不允许需求方直接通过接口访问ES，我们会将各个维度的数据提前算好放入其他类型的数据库中，供业务方使用，此处也不进行展开了。

我们踩过的一些坑

（此处想到什么讲什么了。）

1、ES数据统计查询的时候，同样的查询条件，两次查询出来的数据结果可能会不一样，这是因为副本分片和主分片数据不一致（ES只保证最终一致），ES在写操作的时候有个consistency的参数来控制写入的一致性，具体值为one（primary shard），all（all shard），quorum（default）。

one：要求我们这个写操作，只要有一个primary shard是active活跃可用的，就可以执行
all：要求我们这个写操作，必须所有的primary shard和replica shard都是活跃的，才可以执行这个写操作
quorum：默认的值，要求所有的shard中，必须是大部分的shard都是活跃的，可用的，才可以执行这个写操作

但是就算设置成了all之后，查询还是有不一致的情况，这是使用lucene索引机制带来的refresh问题，彻底解决该问题就势必会增加写入的成本，我们选取了另一种方式：对于会短时间内出现前后两次查询的需求指定从primary shard读。

2、ES查询成功，部分shard失败；这个问题很尴尬，因为我在前期很长时间都没注意到这个问题，发现查询成功后，就直接把结果丢出去了，后来一次ES集群异常，发现查询出来的数据要比正常小很多，不可能是ES主、副本分片数据不一致的问题，才发现时该问题。

3、新增字段的时候，一定要先更新所有已存在的索引的Mapping，再更新template，最后才能发更新后的程序。由于ES集群写操作在默认情况下，Mapping中没有的字段，会被自动识别，而自动识别的字段可能不是我们想要的字段类型，而这个时候想要不断服务的修改，会很复杂。所以一定要在发新的程序之前修改好Mapping、template。

4、有时候为了提高ES集群的性能，我们会定期的手工做一些段合并，此时要注意设置段合并的线程数，防止影响到正常业务。

5、监控ES的慢查询，虽然ES集群是分布式的，但是一样会由于过度的慢查询而打爆集群的情况。

6、做好ES集群的监控很重要，网上有很多教程，此处也不再重述了。