Elasticsearch 入门及在大数据场景下的应用
开篇
Elasticsearch，简称 ES，目前最火的搜索引擎，底层使用开源库 Lucene，拥有丰富的 REST API，开箱即用。分布式的数据存储、倒排索引等设计，使其可以快速存储、搜索、分析海量数据。
当你开始接触 Elasticsearch 并开始使用到业务中时，你会发现对于 Elasticsearch 我们有太多的东西需要去学习，其入门较为简单，想要深入却是挺难的。好在 Elasticsearch 本身迭代更新还是挺快的，官网资料也挺详细，各种玩法也趋于成熟，所以对于初学者来说项目接入 Elasticsearch，也不需要过多的担心实现难度问题。
本次分享将会从以下几点来讲解 ES：
-
ES 安装
-
ES 的基本操作
-
ES 的核心机制
-
ES 在传统数据库无法满足多种条件花式查询情况下的大数据场景应用
-
一些踩坑经历
本篇文章从 ES 的安装、操作到 ES 的应用，由浅及深，适用于初学 ES 的用户。由于篇幅原因，本篇文章会尽量讲清楚文章的核心知识点，对于文章中有不懂的地方可以随时交流，希望可以帮助到大家。
ES 安装
环境准备
ES 的安装环境很简单，只需要提前准备好 Java 环境（jdk1.8），和 ES 的安装包即可，jdk1.8 的安装在此省略，我们直接进入 ES 的安装过程。
下载 ES 安装包
curl -O -L https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.3.2.tar.gz
解压文件
tar -zxvf elasticsearch-5.3.2.tar.gz
修改配置文件
#es集群名称
cluster.name: cluster-es-test
#节点名称，另外两个可以为node-1,node-2
node.name: master-1
#索引数据存储位置，注意给es账户授权
path.data: /usr/local/es/data
#日志文件存储位置，注意给es账户授权
path.logs: /var/log/es
#绑定的ip地址
network.host: 0.0.0.0
#设置对外服务的http端口，默认为9200
http.port: 9200
#设置节点间交互的tcp端口,默认是9300
transport.tcp.port: 9300
#静态设置设置主机列表，每个值应采用host：port或host的形式（其中port默认为设置transport.profiles.default.port，如果未设置则返回transport.tcp.port）
discovery.zen.ping.unicast.hosts: ["host1", "host2", "host3:9300"]
#指定该节点是否有资格被选举成为master节点(这里的true，不代表是master节点，只是有master节点的选举资格)
node.master: true
#允许该节点存储数据(默认开启)
node.data: false
#注意如果需要装head插件，需要解决跨域问题
http.cors.allow-origin: "*"
创建 ES 用户
groupadd esg
useradd esu -g esg
chown -R esu:esg /home/es/elasticsearch-5.3.2
调整 jvm 参数
vim /home/es/elasticsearch-5.3.2/config/jvm.options
-Xms1g
-Xmx1g
官方建议内存设置低于系统内存的一半，不要超过 32G（为什么不要超过 32G 可以翻阅一下资料，这里就不展开了），假如物理机器有 128G，可以在机器上面运行两个 ES 实例。
启动所有节点
/home/es/elasticsearch-5.3.2/bin/elasticsearch -d
验证
http://master1:9200
本章小结
本章主要以 master 节点来介绍 ES 的安装过程，数据节点的安装主要在于配置的不同，启动的时候可能会有一些失败情况，注意观察配置文件目录里面的日志，在此不做展开，有关这方面的问题也可以随时交流。
ES 的基本操作
本篇主要讲述 ES 相对于传统数据库的增删改查操作。
基本概念介绍及与传统关系型数据库类比
拿 MySQL 数据库进行对比，数据库相当于 ES 中的索引（Indices），数据库中的表相当于 ES 中的类型（types），而表中的每一行相当于一个类型里面文档（Documents），而表中的每一列相当于文档中的每个字段（Fields）
Relational DB -> Databases -> Tables -> Rows -> Columns
Elasticsearch -> Indices(数据库)-> Types(表)-> Documents(行)-> Fields(字段)
创建索引
PUT /company?pretty
返回值
{
"acknowledged": true,
"shards_acknowledged": true
}
插入一条数据
PUT /company/user/1?pretty #“1” 为es中索引中的_id,在索引中唯一表示一条数据
{
"name": "张三",
"age": 12,
"sex": 1
}
返回值
{
"_index": "company",
"_type": "user",
"_id": "1",
"_version": 1,
"result": "created",
"_shards":{
"total": 2,
"successful": 1,
"failed": 0
},
"created": true
}
修改数据
PUT /company/user/1?pretty
{
"name": "李四", #名称做修改
"age": 12,
"sex": 1
}
返回值
{
"_index": "company",
"_type": "user",
"_id": "1",
"_version": 2, #版本号变更为2
"result": "updated",
"_shards":{
"total": 2,
"successful": 1,
"failed": 0
},
"created": false
}
可以看到的是文档的版本号变更为 2， ES 中更新数据只能根据索引中的 _id 来更新数据，在其内部，其实是先删除旧的文档，再添加新的文档。
删除文档
DELETE /company/user/1?pretty
返回值
{
"found": true,
"_index": "company",
"_type": "user",
"_id": "1",
"_version": 3, #版本号变更为3
"result": "deleted",
"_shards":{
"total": 2,
"successful": 1,
"failed": 0
}
}
查询数据
GET /company/user/1?pretty
返回值
{
"_index": "company",
"_type": "user",
"_id": "1",
"_version": 4, #再次写入该数据后查询版本号已变更为4
"found": true,
"_source":{
"name": "李四",
"age": 12,
"sex": 1
}
}
Query DSL
DSL：Domain Specified Language，特定领域的语言
http request body：请求体，可以用 json 的格式来构建查询语法，比较方便，可以构建各种复杂的语法，在此不做扩展了。
本章小结
本章主要对于 ES 的增删改查举列说明了一下，主要是帮助大家理解 ES 的一些操作，实际应用中，会根据实际情况做响应的优化，每种操作可能不仅仅这样的简单，在后面的实际应用中，会有响应的例子。
ES 的核心机制
ES 在面向用户的时候，操作起来非常简单，这是因为 ES 内部极力的影藏了其分布式系统复杂性，下面简单的介绍 ES 的一些核心机制。
-
主备机制：和其他的分布式系统一样，ES 也会采取主备机制来保证一些意外情况下数据的完整性，即es中索引的每个分片都会存在主索引和副本索引。当然副本索引也能够提供查询，也极大的提高了 ES 的吞吐量。
-
容错机制：当 ES 的 master 节点丢失的时候（此时集群会变为红色），会从有资格成为主节点的机器中选取一个节点作为新的主节点（此时集群内丢失部分 shard 分片，集群变为黄色），之后将丢失主分片的副本分片升级为主分片，再将这些分片 copy 一份到其他的数据节点上（其中可能会因为数据的不平衡发生数据偏移），之后集群会变为正常状态（绿色）。
-
并发控制：通过 version 版本号来做乐观锁并发控制。系统会检查传递给索引请求的版本号是否大于当前存储的文档的版本，如果大于则重建索引并使用传入的版本号作为新的 version 值，如果提供的值小于或等于存储文档的版本号，则会发生版本冲突，索引操作将失败。在类似的问题处理中，最简单的方法就是使用每次更新数据的时间戳来做版本号。
-
路由机制：ES 通过路由算法（ shard = hash(routing) % number_of_primary_shards）来确定 document 当前属于哪个分片，默认的 routing 值为 document 的 _id，通常我们也会指定别的 Field 来作为 ES 的 routing 值，来提高 ES 查询的性能，但是可能会引起数据的偏移，出现热节点现象。所以 routing 值在选取时需要做充分的评估。
ES 在传统数据库无法满足多种条件花式查询情况下的大数据场景应用
背景
随着数据的增长，传统的数据库很难支撑现有的业务：
-
各种场景的数据查询、统计，导致数据库必须加入各种字段的索引，大大增加了核心链路插入数据所需要的成本，严重影响了核心链路的并发度；大量的索引同时占用了很大的磁盘空间，也给线上 ddl 带来的更大的风险；
-
很多场景没有办法或者很难实现，如：分页、排序、分组查询。大体量数据的关系型数据库设计者必然要考虑水平分库，此时数据在这些操作上面将会异常麻烦。此时我们引入了 ES，来处理允许一定延迟的数据查询、统计的业务。
当传统数据库面对这种大体量数据查询而感到无力的时候，我们可以使用 ES 来处理这种业务
数据存储
在聊数据存储的时候，我先抛出几个问题：
如大家所知的那样，ES 不支持事务，其会利用 _version (版本号)的方式来确保应用中相互冲突的变更不会导致数据丢失，那么我们是如何存储我们的数据，数据结构是什么样子，如何保证数据的时效性、完整性和一致性的呢？
数据结构
首先需要合适的分片数和副本数。目前官方推荐的分片的大小为 20-40G，一般个人会控制在 30G 以内，主副本分片数条件允许的话建议等于节点数。
网上有很多关于如何规划分片数的文章，本人感觉可以作为参考，在机器性能、数据量的大小、使用场景等等的不同，分片数量最好可以通过压测或者线上实际流量来做调整。
我们会尽量减少我们所需要的字段，做到够用就好，mapping 设置方面：设置"_all"为 false，String 类型"index"尽量设置为不分词（"not_analyzed"，根据需要设置 analyzed），如商家名称这类 String 类型字段只存储索引结构，不存储原始文档（后面会聊到如何拿到原始文档）。
起初我们在建索引的时候，我们是尽量冗余数据库中的所有字段整合成一张宽表，导致每天的索引数据量很大，而上面大部分的字段都是不需要的，磁盘利用率很低，而用于该集群的都是 ssd 盘，整体的磁盘容量并不高，常常由于磁盘存不下，而需要添加机器，导致大量的资源浪费。另一方面，尽量的减少我们所需要的字段，这也需要我们支持一个额外的能力，万一需要添加某个字段的时候，我们需要在需求上线之前迅速将历史数据补齐这个字段，同时不影响线上。（我们现在可以一个晚上重刷我们需要周期内的历史数据）。
"mappings": {
"index_type_name": {
"_all": {
"enabled": false
},
"_source": {
"excludes": ["shop_name"] #部分字段只存储其索引部分，不存储原始文档（
},
"properties": {
"order_id": {
"type": "long"
},
"shop_name": {
"type": "string",
"index": "not_analyzed" #提供该字段和关系型数据库like关键字一样功能的查询不需要分词
}
...
}
}
}
以一天为一个索引（根据业务场景，因为我们的业务场景大部分要的都是某天的数据），这也为我们根据实际线上流量调整我们分片、副本数量提供了方便，修改完索引的模板（"_template"）之后，第二天会自动生效，而查询多天不同数量分片的索引的联合查询不会影响查询结果。
注意：分片数一旦固定，那么一个已存在的索引是没有办法直接修改分片数，一般会采取 reindex 的方式重建索引.
尽量避免 Nested 数据类型。每一个 nested 将会作为一个隐藏的单独文本建立索引，虽然官网上说在查询的时候将根文本和 nested 文档拼接是很快的，就跟把他们当成一个单独的文本一样的快。但是其实还是有一部分的额外的消耗，尤其是在用 Nested 中的某个字段作为 aggs 聚合的时候。如果真的需要放入数组类型的数据，可以根据实际需求，转化为一个字段，直接建在主数据上面。
如：我们现在有一个索引，里面有某个学校每天每个学生的学习、生活情况，每个学生每天会产生一条数据。现在我们想统计每个班级某天 中午吃饭时间小于 10 分钟的人数、以及一天在校的用餐次数，我们可以设计一个 nested Objects 数据结构来存储一天三餐的用餐情况，也可以在主数据上添加四个字段：早上吃饭所花时间，中午所花时间、晚上吃饭所花时间，三餐在校用餐次数，这样就可以直接对着这四个字段进行数据统计。
尽量减少 script line 的使用。同样的道理，我们可以预先将需要用 script line 的中间值先存到主数据上面。避免查询、统计时候的额外消耗。
数据如何存储
image
考虑在不影响已有的业务情况下，我们采取解析数据落库产生的 binlog 日志来建索引（binlog 日志公司有一套解决方案，不一定非要使用 binlog 日志，业务状态变更发送的mq消息也是可以的，另外阿里有开源的 Canal，用来做 binlog 日志解析的），使其与正常业务解耦。
此时我们不会直接拿这条数据插入 ES，因为数据的状态变化在同一个时刻可能会发生多次，每次的数据插入不一定是数据库当前的最新状态，而且无论是 binlog 日志、还是状态变化 MQ 消息都只是涵盖了部分数据，如果要数据在发送 MQ 消息的时候，把建索引所需要的数据补齐之后发送给你，对于的原有业务来说，会面临经常修改 MQ 消息结构的问题，这已经违背了我们要使其与正常业务解耦的初衷，所以我们在收到这条数据变更的时候，会通过 soa 接口的方式反查当前宽表数据，这样补齐宽表数据之后再写索引，这样我们就可以拿到我们想要的任何最新数据。
通过 ES 建索引的 **bulk api **减少与 ES 集群的交互次数，提高数据写入的吞吐量。
同一条数据，在同一个时刻可能会在机器 A 和机器 B 中同时发生写索引操作，机器 A 查询到的是旧数据，机器 B 查询到了新数据，但是写入索引的时候机器 B 先写入 ES 集群，机器 A 后写入集群，导致数据错误。解决方案：每条数据写入的时候，添加一个分布式锁，相同唯一建的数据在同一个时刻只能有一条发生写索引的动作，没有获得分布式锁消息，丢入延迟队列，下次再消费（由于存在 soa 接口重新反查当前宽表数据，所以也不需要担心消息乱序消费的问题）。
数据的补偿（此处就不展开了）。
数据的查询和统计
数据查询
前面我们讲述了我们 ES 中的索引结构遵循的一些原则，其中有一条是，我们只存储字段的索引部分来提供查询，不会在 ES 中存储原始文档来提供输出，那么当查询方需要该字段的信息的时候，我们需要怎么做？其实这是一个 ES 集群的定位问题，我们的 ES 集群仅仅是用来丰富数据查询、支持数据统计的功能，我们并不支持数据的实际存储，我们存储的仅仅只是每个字段的索引而已，通过每个字段的索引支持各种各样的数据查询、数据统计，如果需要查询数据的详细信息，我们可以通过 ES 查询得到数据的唯一键后，再通过 soa 接口来反查该数据，之后吐给需求方，我们会将这个步骤包掉，需求方无感知，且返回的数据只是将原有的 soa 查询接口的返回数据包了一层，让其他方感知到的是原 soa 接口的升级版接口，尽可能减少其他方的接入成本。
[图片上传失败...(image-e47026-1561369954378)]
基于 ES 的数据的统计
ES 在做数据统计的时候往往会很消耗 ES 集群的资源，所以我们通常不允许需求方直接通过接口访问 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 问题。
解决该问题需要将 write consistency 设置成为 all，replication 是 sync 模式(默认)，每次写入数据时候需要手工 refresh。如果是写入很大的应用不建议这样去做。我们选取了另一种方式：对于会短时间内出现前后两次查询，且需要保证两次查询出来的数据强一致性的需求指定从 primary shard 读。
ES 查询成功，部分 shard 失败
这个问题很尴尬，因为我在前期很长时间都没注意到这个问题，发现查询返回成功后，就直接把结果丢出去了，并没有注意到下面还有失败 shards 数量，后来一次 ES 集群异常，发现查询出来的数据要比正常值普遍要小很多，不可能是 ES 主、副本分片数据不一致的问题，才发现是该问题。
新增字段
新增字段的时候，一定要先更新所有已存在的索引的 Mapping，再更新 template，最后才能发布更新后的程序。
由于 ES 集群写操作在默认情况下，Mapping 中没有的字段，会被自动识别，而自动识别的字段可能不是我们想要的字段类型，而这个时候想要不断服务的修改，会很复杂。所以一定要在发新的程序之前修改好Mapping、template。
定期进行段合并，导致线上大量报错
有时候为了提高 ES 集群的性能，我们会定期的手工做一些段合并，但是段合并的计算量庞大，而且会吃掉大量的磁盘 IO，此时要注意设置段合并的线程数，避开业务高峰期，防止影响到正常业务。
慢查询拖垮 ES 集群
我们解决这个问题主要有两点：
- 慢查询需要有监控措施；
- 提供给 ES 的查询接口限流、降级措施需要做好。
前期设计工作一定要做好
尤其是对于 mapping 的设计，一旦设计的不合理，后期再想修改会很费劲，需要重新再长出一套同样的东西供调用方使用，然后才能彻底下掉老逻辑，周期会很长。
做好 ES 集群的监控很重要，否则出现问题，就跟瞎子一样，很难定位问题。
小结
以上，便是本篇的全部内容，Elasticsearch 里面可以讲的有太多太多了，每个小点拿出来可能都能说上一番。本篇文章较浅，如果有您不满意的的地方，希望可以多多指点，同时也希望本篇文章能够对您有所收获。
本文首发于 GitChat，未经授权不得转载，转载需与 GitChat 联系。
