让Elasticsearch飞起来！百亿级实时查询优化实战

雨点打透心脏的1/2处 2021-11-17 22:20 376阅读 0赞

最近的一个项目是风控过程数据实时统计分析和聚合的一个 OLAP 分析监控平台，日流量峰值在 10 到 12 亿上下，每年数据约 4000 亿条，占用空间大概 200T。

面对这样一个数据量级的需求，我们的数据如何存储和实现实时查询将是一个严峻的挑战。

经过对 Elasticsearch 多方调研和超过几百亿条数据的插入和聚合查询的验证之后，我们总结出以下几种能够有效提升性能和解决这一问题的方案：

*  **集群规划**
 *  **存储策略**
 *  **索引拆分**
 *  **压缩**
 *  **冷热分区等**

本文所使用的 Elasticsearch 版本为 5.3.3。

什么是时序索引？其主要特点体现在如下两个方面：

*  **存，**以时间为轴，数据只有增加，没有变更，并且必须包含 timestamp（日期时间，名称随意）字段。
    
    其作用和意义要大于数据的 id 字段，常见的数据比如我们通常要记录的操作日志、用户行为日志、或股市行情数据、服务器 CPU、内存、网络的使用率等。
 *  **取，**一定是以时间范围为第一过滤条件，然后是其他查询条件，比如近一天、一周、本月等等，然后在这个范围内进行二次过滤。
    
    比如性别或地域等，查询结果中比较关注的是每条数据和 timestamp 字段具体发生的时间点，而非 id。

此类数据一般用于 OLAP、监控分析等场景。

集群部署规划

我们都知道在 Elasticsearch（下称 ES）集群中有两个主要角色：Master Node 和 Data Node，其他如 Tribe Node 等节点可根据业务需要另行设立。

为了让集群有更好的性能表现，我们应该对这两个角色有一个更好的规划，在 Nodes 之间做读取分离，保证集群的稳定性和快速响应，在大规模的数据存储和查询的压力之下能够坦然面对，各自愉快的协作。

**Master Nodes**

Master Node，整个集群的管理者，负有对 index 的管理、shards 的分配，以及整个集群拓扑信息的管理等功能。

众所周知，Master Node 可以通过 Data Node 兼任，但是，如果对群集规模和稳定要求很高的话，就要职责分离，Master Node 推荐独立，它的状态关乎整个集群的存活。

Master 的配置：

node.master: true 
    node.data: false 
    node.ingest: false

这样 Master 不参与 I、O，从数据的搜索和索引操作中解脱出来，专门负责集群的管理工作，因此 Master Node 的节点配置可以相对低一些。

另外防止 ES 集群 split brain（脑裂），合理配置 discovery.zen.minimum\_master\_nodes 参数，官方推荐 master-eligible nodes / 2 + 1 向下取整的个数。

这个参数决定选举 Master 的 Node 个数，太小容易发生“脑裂”，可能会出现多个 Master，太大 Master 将无法选举。

更多 Master 选举相关内容请参考：

https://www.elastic.co/guide/en/elasticsearch/reference/5.3/modules-discovery-zen.html#master-election

**Data Nodes**

Data Node 是数据的承载者，对索引的数据存储、查询、聚合等操作提供支持。

这些操作严重消耗系统的 CPU、内存、IO 等资源，因此，应该把好的资源分配给 Data Node，因为它们是真正干累活的角色，同样 Data Node 也不兼任 Master 的功能。

Data 的配置：

node.master: false 
    node.data: true 
    node.ingest: false

**Coordinating Only Nodes**

ES 本身是一个分布式的计算集群，每个 Node 都可以响应用户的请求，包括 Master Node、Data Node，它们都有完整的 Cluster State 信息。

正如我们知道的一样，在某个 Node 收到用户请求的时候，会将请求转发到集群中所有索引相关的 Node 上，之后将每个 Node 的计算结果合并后返回给请求方。

我们暂且将这个 Node 称为查询节点，整个过程跟分布式数据库原理类似。那问题来了，这个查询节点如果在并发和数据量比较大的情况下，由于数据的聚合可能会让内存和网络出现瓶颈。

因此，在职责分离指导思想的前提下，这些操作我们也应该从这些角色中剥离出来，官方称它是 Coordinating Nodes，只负责路由用户的请求，包括读、写等操作，对内存、网络和 CPU 要求比较高。

本质上，Coordinating Only Nodes 可以笼统的理解为是一个负载均衡器，或者反向代理，只负责读，本身不写数据，它的配置是：

node.master: false 
    node.data: false 
    node.ingest: false 
    search.remote.connect: false

增加 Coordinating Nodes 的数量可以提高 API 请求响应的性能，我们也可以针对不同量级的 Index 分配独立的 Coordinating Nodes 来满足请求性能。

那是不是越多越好呢？在一定范围内是肯定的，但凡事有个度，过了负作用就会突显，太多的话会给集群增加负担。

在做 Master 选举的时候会先确保所有 Node 的 Cluster State 是一致的，同步的时候会等待每个 Node 的 Acknowledgement 确认，所以适量分配可以让集群畅快的工作。

search.remote.connect 是禁用跨集群查询，防止在进行集群之间查询时发生二次路由：

https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-cross-cluster-search.html

Routing

类似于分布式数据库中的分片原则，将符合规则的数据存储到同一分片。ES 通过哈希算法来决定数据存储于哪个 Shard：

shard_num = hash(_routing) % num_primary_shards

其中 hash(\_routing) 得出一个数字，然后除以主 Shards 的数量得到一个余数，余数的范围是 0 到 number\_of\_primary\_shards - 1，这个数字就是文档所在的 Shard。

Routing 默认是 id 值，当然可以自定义，合理指定 Routing 能够大幅提升查询效率，Routing 支持 GET、Delete、Update、Post、Put 等操作。

如：

PUT my_index/my_type/1?routing=user1
    {
      "title": "This is a document"
    }
    
    GET my_index/my_type/1?routing=user1

不指定 Routing 的查询过程：

简单的来说，一个查询请求过来以后会查询每个 Shard，然后做结果聚合，总的时间大概就是所有 Shard 查询所消耗的时间之和。

指定 Routing 以后：

会根据 Routing 查询特定的一个或多个 Shard，这样就大大减少了查询时间，提高了查询效率。

当然，如何设置 Routing 是一个难点，需要一点技巧，要根据业务特点合理组合 Routing 的值，来划分 Shard 的存储，最终保持数据量相对均衡。

可以组合几个维度做为 Routing ，有点类似于 Hbase Key，例如不同的业务线加不同的类别，不同的城市和不同的类型等等，如：

*  **\_search?routing=beijing：**按城市。
 *  **\_search?routing=beijing\_user123：**按城市和用户。
 *  **\_search?routing=beijing\_android，shanghai\_android：**按城市和手机类型等。

数据不均衡？假如你的业务在北京、上海的数据远远大于其他二三线城市的数据。

再例如我们的业务场景，A 业务线的数据量级远远大于 B 业务线，有时候很难通过 Routing 指定一个值保证数据在所有 Shards 上均匀分布，会让部分 Shard 变的越来越大，影响查询性能，怎么办？

一种解决办法是单独为这些数据量大的渠道创建独立的 Index，如：

http://localhost:9200/shanghai,beijing,other/_search?routing=android

这样可以根据需要在不同 Index 之间查询，然而每个 Index 中 Shards 的数据可以做到相对均衡。

另一种办法是指定 Index 参数 index.routing\_partition\_size，来解决最终可能产生群集不均衡的问题，指定这个参数后新的算法如下：

shard_num = (hash(_routing) + hash(_id) % routing_partition_size) % num_primary_shards

index.routing\_partition\_size 应具有大于 1 且小于 index.number\_of\_shards 的值。

最终数据会在 routing\_partition\_size 几个 Shard 上均匀存储，是哪个 Shard 取决于 hash(\_id) % routing\_partition\_size 的计算结果。

指定参数 index.routing\_partition\_size 后，索引中的 Mappings 必须指定 \_routing 为 "required": true，另外 Mappings 不支持 parent-child 父子关系。

很多情况下，指定 Routing 后会大幅提升查询性能，毕竟查询的 Shard 只有那么几个，但是如何设置 Routing 是个难题，可根据业务特性巧妙组合。

索引拆分

Index 通过横向扩展 Shards 实现分布式存储，这样可以解决 Index 大数据存储的问题。

但在一个 Index 变的越来越大，单个 Shard 也越来越大，查询和存储的速度也越来越慢。

更重要的是一个 Index 其实是有存储上限的（除非你设置足够多的 Shards 和机器），如官方声明单个 Shard 的文档数不能超过 20 亿（受限于 Lucene index，每个 Shard 是一个 Lucene index）。

考虑到 I、O，针对 Index 每个 Node 的 Shards 数最好不超过 3 个，那面对这样一个庞大的 Index，我们是采用更多的 Shards，还是更多的 Index，我们如何选择？

Index 的 Shards 总量也不宜太多，更多的 Shards 会带来更多的 I、O 开销，其实答案就已经很明确，除非你能接受长时间的查询等待。

Index 拆分的思路很简单，时序索引有一个好处就是只有增加，没有变更，按时间累积，天然对索引的拆分友好支持，可以按照时间和数据量做任意时间段的拆分。

ES 提供的 Rollover Api + Index Template 可以很便捷和友好的实现 Index 的拆分工作，把单个 index docs 数量控制在百亿内，也就是一个 Index 默认 5 个 Shards 左右即可，保证查询的即时响应。

简单介绍一下 Rollover API 和 Index Template 这两个东西，如何实现 index 的拆分。

**Index Template**

我们知道 ES 可以为同一目的或同一类索引创建一个 Index Template，之后创建的索引只要符合匹配规则就会套用这个 Template，不必每次指定 Settings 和 Mappings 等属性。

一个 Index 可以被多个 Template 匹配，那 Settings 和 Mappings 就是多个 Template 合并后的结果。

有冲突通过 Template 的属性"order" : 0 从低到高覆盖（这部分据说会在 ES6 中会做调整，更好的解决 Template 匹配冲突问题）。

示例：

PUT _template/template_1
    {
        "index_patterns" : ["log-*"],
        "order" : 0,
        "settings" : {
            "number_of_shards" : 5
        },
        "aliases" : {
            "alias1" : {}
        }
    }

**Rollover Index**

Rollover Index 可以将现有的索引通过一定的规则，如数据量和时间，索引的命名必须是 logs-000001 这种格式，并指定 aliases，示例：

PUT /logs-000001 
    {
      "aliases": {
        "logs_write": {}
      }
    }
    
    # Add > 1000 documents to logs-000001
    
    POST /logs_write/_rollover 
    {
      "conditions": {
        "max_age":   "7d",
        "max_docs":  1000
      }
    }

先创建索引并指定别名 logs\_write，插入 1000 条数据，然后请求 \_rollover api 并指定拆分规则。

如果索引中的数据大于规则中指定的数据量或者时间过时，新的索引将被创建，索引名称为 logs-000002，并根据规则套用 Index Template，同时别名 logs\_write 也将被变更到 logs-000002。

注意事项：

*  索引命名规则必须如同：logs-000001。
 *  索引必须指定 aliases。
 *  Rollover Index API 调用时才去检查索引是否超出指定规则，不会自动触发，需要手动调用，可以通过 Curator 实现自动化。
 *  如果符合条件会创建新的索引，老索引的数据不会发生变化，如果你已经插入 2000 条，拆分后还是 2000 条。
 *  插入数据时一定要用别名，否则你可能一直在往一个索引里追加数据。

技巧是按日期滚动索引：

PUT /<logs-{now/d}-1>
    {
      "aliases": {
        "logs_write": {}
      }
    }

假如生成的索引名为 logs-2017.04.13-1，如果你在当天执行 Rollover 会生成 logs-2017.04.13-000001，次日的话是 logs-2017.04.14-000001。

这样就会按日期进行切割索引，那如果你想查询 3 天内的数据可以通过日期规则来匹配索引名，如：

GET /<logs-{now/d}-*>,<logs-{now/d-1d}-*>,<logs-{now/d-2d}-*>/_search

到此，我们就可以通过 Index Template 和 Rollover API 实现对 Index 的任意拆分，并按照需要进行任意时间段的合并查询，这样只要你的时间跨度不是很大，查询速度一般可以控制在毫秒级，存储性能也不会遇到瓶颈。

Hot-Warm 架构

冷热架构，为了保证大规模时序索引实时数据分析的时效性，可以根据资源配置不同将 Data Nodes 进行分类形成分层或分组架构。

一部分支持新数据的读写，另一部分仅支持历史数据的存储，存放一些查询发生机率较低的数据。

即 Hot-Warm 架构，对 CPU，磁盘、内存等硬件资源合理的规划和利用，达到性能和效率的最大化。

我们可以通过 ES 的 Shard Allocation Filtering 来实现 Hot-Warm 的架构。

实现思路如下：

*  将 Data Node 根据不同的资源配比打上标签，如：Host、Warm。
 *  定义 2 个时序索引的 Index Template，包括 Hot Template 和 Warm Template，Hot Template 可以多分配一些 Shard 和拥有更好资源的 Hot Node。
 *  用 Hot Template 创建一个 Active Index 名为 active-logs-1，别名 active-logs，支持索引切割。
    
    插入一定数据后，通过 roller over api 将 active-logs 切割，并将切割前的 Index 移动到 Warm Nodes 上，如 active-logs-1，并阻止写入。
 *  通过 Shrinking API 收缩索引 active-logs-1 为 inactive-logs-1，原 Shard 为 5，适当收缩到 2 或 3，可以在 Warm Template 中指定，减少检索的 Shard，使查询更快。
 *  通过 force-merging api 合并 inactive-logs-1 索引每个 Shard 的 Segment，节省存储空间。
 *  删除 active-logs-1。

## ##

**Hot，Warm Nodes**

**Hot Nodes**

拥有好资源的 Data Nodes，如更高性能的 CPU、SSD 磁盘、内存等资源，这些特殊的 Nodes 支持索引所有的读、写操作。

如果你计划以 100 亿为单位来切割 Index，那至少需要三个这样的 Data Nodes，Index 的 Shard 数为 5，每个 Shard 支持 20 亿 Documents 数据的存储。

为这类 Data Nodes 打上标签，以便我们在 Template 中识别和指定，启动参数如下：

./bin/elasticsearch -Enode.attr.box_type=hot

或者配置文件：

node.attr.box_type: hot

**Warm Nodes**

存储只读数据，并且查询量较少，但用于存储长多时间历史数据的 Data Nodes，这类 Nodes 相对 Hot Nodes 较差的硬件配置，根据需求配置稍差的 CPU、机械磁盘和其他硬件资源，至于数量根据需要保留多长时间的数据来配比，同样需要打上标签，方法跟 Hot Nodes 一样，指定为 Warm，box\_type: warm。

**Hot，Warm Template**

**Hot Template**

我们可以通过指定参数"routing.allocation.include.box\_type": "hot"，让所有符合命名规则索引的 Shard 都将被分配到 Hot Nodes 上：

PUT _template/active-logs
    {
      "template": "active-logs-*",
      "settings": {
        "number_of_shards":   5,
        "number_of_replicas": 1,
        "routing.allocation.include.box_type": "hot",
        "routing.allocation.total_shards_per_node": 2
      },
      "aliases": {
        "active-logs":  {}
      }
    }

**Warm Template**

同样符合命名规则索引的 Shard 会被分配到 Warm Nodes 上，我们指定了更少的 Shards 数量和复本数。

注意，这里的复本数为 0，和 best\_compression 级别的压缩，方便做迁移等操作，以及进行一些数据的压缩：

PUT _template/inactive-logs
    {
      "template": "inactive-logs-*",
      "settings": {
        "number_of_shards":   1,
        "number_of_replicas": 0,
        "routing.allocation.include.box_type": "warm",
        "codec": "best_compression"
      }
    }

假如我们已经创建了索引 active-logs-1 ，当然你可以通过 \_bulk API 快速写入测试的数据，然后参考上文中介绍的 Rollover API 进行切割。

**Shrink Index**

Rollover API 切割以后，active-logs-1 将变成一个冷索引，我们将它移动到 Warm Nodes 上。

先将索引置为只读状态，拒绝任何写入操作，然后修改 index.routing.allocation.include.box\_type 属性，ES 会自动移动所有 Shards 到目标 Data Nodes 上：

PUT active-logs-1/_settings
    {
      "index.blocks.write": true,
      "index.routing.allocation.include.box_type": "warm"
    }

Cluster Health API 可以查看迁移状态，完成后进行收缩操作，其实相当于复制出来一个新的索引，旧的索引还存在。

POST active-logs-1/_shrink/inactive-logs-1

我们可以通过 Head 插件查看整个集群索引的变化情况。关于 Shard 的分配请参考：

https://www.elastic.co/guide/en/elasticsearch/reference/current/shard-allocation-filtering.html

**Forcemerge**

到目前为止我们已经实现了索引的冷热分离，和索引的收缩，我们知道每个 Shard 下面由多个 Segment 组成，那 inactive-logs-1 的 Shard 数是 1，但 Segment 还是多个。

这类索引不会在接受写入操作，为了节约空间和改善查询性能，通过 Forcemerge API 将 Segment 适量合并：

PUT inactive-logs-1/_settings
    { "number_of_replicas": 1 }

ES 的 Forcemerge 过程是先创建新的 Segment 删除旧的，所以旧 Segment 的压缩方式 best\_compression 不会在新的 Segment 中生效，新的 Segment 还是默认的压缩方式。

现在 inactive-logs-1 的复本还是 0，如果有需要的话，可以分配新的复本数：

PUT inactive-logs-1/_settings
    { "number_of_replicas": 1 }

最后删除 active-logs-1，因为我们已经为它做了一个查询复本 inactive-logs-1。

**DELETE active-logs-1**

走到这里，我们就已经实现了 Index 的 Hot-Warm 架构，根据业务特点将一段时间的数据放在 Hot Nodes，更多的历史数据存储于 Warm Nodes。

其他优化方案

**索引隔离**

在多条业务线的索引共用一个 ES 集群时会发生流量被独吃独占的情况，因为大家都共用相同的集群资源，流量大的索引会占用大部分计算资源而流量小的也会被拖慢，得不到即时响应，或者说业务流量大的索引可以按天拆分，几个流量小的索引可以按周或月拆分。

这种情况下我们可以将规模大的索引和其他相对小规模的索引独立存储，分开查询或合并查询。

除了 Master Nodes 以外，Data Nodes 和 Coordinating Nodes 都可以独立使用（其实如果每个索引的量都特别大也应该采用这种架构），还有一个好处是对方的某个 Node 挂掉，自己不受影响。

同样利用 ES 支持 Shard Allocation Filtering 功能来实现索引的资源独立分配，先将 Nodes 进行打标签，划分区域，类似于 Hot-Warm 架构：

node.attr.zone=zone_a、node.attr.zone=zone_b

或者：

node.attr.zone =zone_hot_a、node.attr.zone=zone_hot_b

等打标签的方式来区别对应不同的索引，然后在各自的 Index Template 中指定不同的 node.attr.zone 即可。

如"index.routing.allocation.include.zone" : "zone\_a,zone\_hot\_a"，

或者排除法"index.routing.allocation.exclude.size": "zone\_hot\_b"

分配到 zone\_hot\_b 以外的所有 Data Nodes 上。

更多用法可以参考 Hot-Warm 架构，或 shard-allocation-filtering：

https://www.elastic.co/guide/en/elasticsearch/reference/current/shard-allocation-filtering.html

**跨数据中心**

如果你的业务遍布全国各地，四海八荒，如果你数据要存储到多个机房，如果你的 Index 有几万个甚至更多（ Index 特别多，集群庞大会导致 Cluster State 信息量特别大，因为 Cluster State 包含了所有 Shard、Index、Node 等所有相关信息，它存储在每个 Node 上，这些数据发生变化都会实时同步到所有 Node 上，当这个数据很大的时候会对集群的性能造成影响）。

这些情况下我们会考虑部署多个 ES Cluster，那我们将如何解决跨集群查询的问题呢？

目前 ES 针对跨集群操作提供了两种方案 Tribe Node 和 Cross Cluster Search。

**Tribe Node**

需要一个独立的 Node 节点，加入其他 ES Cluster，用法有点类似于 Coordinating Only Node。

所不同的是 Tribe 是针对多个 ES 集群之间的所有节点，Tribe Node 收到请求广播到相关集群中所有节点，将结果合并处理后返回。

表面上看起来 Tribe Node 将多个集群串连成了一个整体，遇到同名 Index 发生冲突，会选择其中一个 Index，也可以指定：

tribe:
      on_conflict: prefer_t1
      t1:
        cluster.name: us-cluster
        discovery.zen.ping.multicast.enabled: false
        discovery.zen.ping.unicast.hosts: ['usm1','usm2','usm3']
      t2:
        cluster.name: eu-cluster
        discovery.zen.ping.multicast.enabled: false
        discovery.zen.ping.unicast.hosts: ['eum1','eum2','eum3']

**Cross Cluster Search**

Cross Cluster Search 可以让集群中的任意一个节点联合查询其他集群中的数据， 通过配置 elasticsearch.yml 或者 API 来启用这个功能，API 示例：

PUT _cluster/settings
    {
      "persistent": {
        "search": {
          "remote": {
            "cluster_one": {
              "seeds": [
                "127.0.0.1:9300"
              ]
        ...
            }
          }
        }
      }
    }

提交以后整个集群所有节点都会生效，都可以做为代理去做跨集群联合查询，不过我们最好还是通过 Coordinating Only Nodes 去发起请求。

POST /cluster_one:decision,decision/_search
    {
        "match_all": {}
    }

对集群 cluster\_one 和本集群中名为 Decision 的索引联合查询。

目前这个功能还在测试阶段，但未来可能会取代 Tribe Node，之间的大的差异是 Tribe Node 需要设置独立的节点，而 Cross Cluster Search 不需要，集群中的任意一个节点都可以兼任。

比如可以用我们的 Coordinating Only Nodes 做为联合查询节点，另一个优点是配置是动态的，不需要重启节点。

实际上可以理解为是一个 ES 集群之间特定的动态代理工具，支持所有操作，包括 Index 的创建和修改，并且通过 Namespace 对 Index 进行隔离，也解决了 Tribe Node 之 Index 名称冲突的问题。

小结

我们在文中介绍了几种方案用来解决时序索引的海量数据存储和查询的问题，根据业务特点和使用场景来单独或组合使用能够发挥出意想不到的效果。

特别是 Nodes 之间的读写分离、索引拆分、Hot-Warm 等方案的组合应用对索引的查询和存储性能有显著的提升。

另外 Routing 在新版本中增加了 routing\_partition\_size，解决了 Shard 难以均衡的问题。

如果你的索引 Mapping 中没有 parent-child 关联关系可以考虑使用，对查询的性能提升非常有效。

文章来源：[https://blog.csdn.net/tkkzc3E6s4Ou4/article/details/88968378][https_blog.csdn.net_tkkzc3E6s4Ou4_article_details_88968378]