【ES】ElasticSearch 深入分片

Posted 2023-03-16

参考技术A

@[toc]

分片是 Elasticsearch 在集群中分发数据的关键。

把分片想象成数据的容器。文档存储在分片中，然后分片分配到集群中的节点上。当集群扩容或缩小，Elasticsearch 将会自动在节点间迁移分片，以使集群保持平衡。

一个分片(shard)是一个最小级别“工作单元(worker unit)”，它只是保存了索引中所有数据的一部分。

这类似于 mysql 的分库分表，只不过 Mysql 分库分表需要借助第三方组件而 ES 内部自身实现了此功能。

分片可以是 主分片(primary shard) 或者是 复制分片(replica shard) 。

在集群中唯一一个空节点上创建一个叫做 blogs 的索引。默认情况下，一个索引被分配 5 个主分片，下面只分配 3 个主分片和一个复制分片（每个主分片都有一个复制分片）：

在一个多分片的索引中写入数据时，通过路由来确定具体写入哪一个分片中，大致路由过程如下：

routing 是一个可变值，默认是文档的 _id ，也可以设置成一个自定义的值。routing 通过 hash 函数生成一个数字，然后这个数字再除以 number_of_primary_shards （主分片的数量）后得到余数 。这个在 0 到 number_of_primary_shards 之间的余数，就是所寻求的文档所在分片的位置。

这解释了为什么要在创建索引的时候就确定好主分片的数量并且永远不会改变这个数量： 因为如果数量变化了，那么所有之前路由的值都会无效，文档也再也找不到了 。

索引中的每个文档属于一个单独的主分片，所以 主分片的数量决定了索引最多能存储多少数据 （实际的数量取决于数据、硬件和应用场景）。

复制分片只是主分片的一个副本，它可以 防止硬件故障导致的数据丢失，同时可以提供读请求，比如搜索或者从别的 shard 取回文档 。

每个主分片都有一个或多个副本分片，当主分片异常时，副本可以提供数据的查询等操作。主分片和对应的副本分片是不会在同一个节点上的，所以副本分片数的最大值是 n -1（其中 n 为节点数）。

当索引创建完成的时候，主分片的数量就固定了，但是复制分片的数量可以随时调整，根据需求扩大或者缩小规模。如把复制分片的数量从原来的 1 增加到 2 ：

分片本身就是一个完整的搜索引擎，它可以使用单一节点的所有资源。 主分片或者复制分片都可以处理读请求——搜索或文档检索，所以数据的冗余越多，能处理的搜索吞吐量就越大。

对文档的新建、索引和删除请求都是写操作，必须在主分片上面完成之后才能被复制到相关的副本分片，ES 为了提高写入的能力这个过程是并发写的，同时为了解决并发写的过程中数据冲突的问题，ES 通过乐观锁的方式控制，每个文档都有一个 _version （版本）号，当文档被修改时版本号递增。一旦所有的副本分片都报告写成功才会向协调节点报告成功，协调节点向客户端报告成功。

ES 集群中每个节点通过路由都知道集群中的文档的存放位置，所以每个节点都有处理读写请求的能力。

在一个写请求被发送到某个节点后，该节点即为协调节点，协调节点会根据路由公式计算出需要写到哪个分片上，再将请求转发到该分片的主分片节点上。假设 shard = hash(routing) % 4 = 0 ，则过程大致如下：

写入磁盘的倒排索引是不可变的，优势主要表现在：

当然，不可变的索引有它的缺点：

在全文检索的早些时候，会为整个文档集合建立一个大索引，并且写入磁盘。只有新的索引准备好了，它就会替代旧的索引，最近的修改才可以被检索。这无疑是低效的。

因为索引的不可变性带来的好处，那如何在保持不可变同时更新倒排索引？

答案是，使用多个索引。 不是重写整个倒排索引，而是增加额外的索引反映最近的变化。 每个倒排索引都可以按顺序查询，从最老的开始，最后把结果聚合。

这就引入了 段 （segment） ：

分片下的索引文件被拆分为多个子文件，每个子文件叫作 段 ， 每一个段本身都是一个倒排索引，并且段具有不变性，一旦索引的数据被写入硬盘，就不可再修改。

段被 写入到磁盘 后会生成一个 提交点 ，提交点是一个用来记录所有提交后段信息的文件。一个段一旦拥有了提交点，就说明这个段只有读的权限，失去了写的权限。相反，当段在内存中时，就只有写的权限，而不具备读数据的权限，意味着不能被检索。

在 Lucene 中的索引（Lucene 索引是 ES 中的分片，ES 中的索引是分片的集合）指的是段的集合，再加上提交点(commit point)，如下图：

在底层采用了分段的存储模式，使它在读写时几乎完全避免了锁的出现，大大提升了读写性能。

索引文件分段存储并且不可修改 ，那么新增、更新和删除如何处理呢？

ES 是怎么做到 近实时 全文搜索？

磁盘是瓶颈。提交一个新的段到磁盘需要 fsync 操作，确保段被物理地写入磁盘，即时电源失效也不会丢失数据。但是 fsync 是昂贵的，严重影响性能，当写数据量大的时候会造成 ES 停顿卡死，查询也无法做到快速响应。

所以 fsync 不能在每个文档被索引的时就触发，需要一种更轻量级的方式使新的文档可以被搜索，这意味移除 fsync 。

为了提升写的性能，ES 没有每新增一条数据就增加一个段到磁盘上，而是采用 延迟写 的策略。

每当有新增的数据时，就将其先写入到内存中，在内存和磁盘之间是文件系统缓存，当达到默认的时间（1秒钟）或者内存的数据达到一定量时，会触发一次刷新（Refresh），将内存中的数据生成到一个新的段上并缓存到文件缓存系统 上，稍后再被刷新到磁盘中并生成提交点 。

这里的内存使用的是ES的JVM内存，而文件缓存系统使用的是操作系统的内存。新的数据会继续的被写入内存，但内存中的数据并不是以段的形式存储的，因此不能提供检索功能。由内存刷新到文件缓存系统的时候会生成了新的段，并将段打开以供搜索使用，而不需要等到被刷新到磁盘。

在 Elasticsearch 中，这种写入和打开一个新段的轻量的过程叫做 refresh （即内存刷新到文件缓存系统）。默认情况下每个分片会每秒自动刷新一次。 这就是为什么说 Elasticsearch 是近实时的搜索了：文档的改动不会立即被搜索，但是会在一秒内可见。

也可以手动触发 refresh。 POST /_refresh 刷新所有索引， POST /index/_refresh 刷新指定的索引：

没用 fsync 同步文件系统缓存到磁盘，不能确保电源失效，甚至正常退出应用后，数据的安全。为了 ES 的可靠性，需要确保变更持久化到磁盘。

虽然通过定时 Refresh 获得近实时的搜索，但是 Refresh 只是将数据挪到文件缓存系统，文件缓存系统也是内存空间，属于操作系统的内存，只要是内存都存在断电或异常情况下丢失数据的危险。

为了避免丢失数据，Elasticsearch添加了 事务日志（Translog） ，事务日志记录了所有还没有持久化到磁盘的数据。

有了事务日志，过程现在如下：

事务日志记录了没有 flush 到硬盘的所有操作。当故障重启后，ES 会用最近一次提交点从硬盘恢复所有已知的段，并且从日志里恢复所有的操作。

在 ES 中，进行一次提交并删除事务日志的操作叫做 flush 。分片每 30 分钟，或事务日志过大会进行一次 flush 操作。 flush API 也可用来进行一次手动 flush ， POST/ _flush 针对所有索引有效， POST /index/_flush 则指定的索引：

通常很少需要手动 flush ，通常自动的就够了。

总体的流程大致如下：

由于自动刷新流程每秒会创建一个新的段 ，这样会导致短时间内的段数量暴增。而段数目太多会带来较大的麻烦。每一个段都会消耗文件句柄、内存和 cpu 运行周期。更重要的是，每个搜索请求都必须轮流检查每个段然后合并查询结果，所以段越多，搜索也就越慢。

ES 通过后台合并段解决这个问题。小段被合并成大段，再合并成更大的段。这时旧的文档从文件系统删除的时候，旧的段不会再复制到更大的新段中。合并的过程中不会中断索引和搜索。

段合并在进行索引和搜索时会自动进行，合并进程选择一小部分大小相似的段，并且在后台将它们合并到更大的段中，这些段既可以是未提交的也可以是已提交的。

合并结束后老的段会被删除，新的段被 flush 到磁盘，同时写入一个包含新段且排除旧的和较小的段的新提交点，新的段被打开可以用来搜索。

1. 全文搜索引擎Elasticsearch，这篇文章给讲透了
2.ElasticSearch 权威指南》

日志分析 第六章 安装elasticsearch

在这里，以两台es集群为例。

es集群健康状况有三种状态，这里我们搭建的es集群，只要两台不同时挂掉，数据不会丢失。

green	所有主要分片和复制分片都可用
yellow	所有主要分片可用，但不是所有复制分片都可用
red	不是所有的主要分片都可用

 

 

 

 

举个例子：

• 比如说现在集群节点es1位主节点，es2位复制分片节点，默认情况下，两台es都接收logstash传过来的日志，是负载均衡的。
• 如果es1宕掉，es2会被提升为主节点，只有es2接收logstash传来得日志数据，同时整个集群状态由green转为red。
• 修复es1，es1会作为分片节点加入集群中，es1加入到集群中后，会进行修复，es2中新增的数据同步到es1分片节点，同步过程整个集群健康状态转为yellow。
• 当同步完成，集群状态转为green。

安装elasticsearch

官网：https://www.elastic.co/products/elasticsearch

# tar xf elasticsearch-2.3.5.tar.gz -C /usr/local/app/
# ln -sv /usr/local/app/elasticsearch-2.3.5 /usr/local/elasticsearch
# chown –R nobody.nobody /usr/local/app/elasticsearch-2.3.5/
# cd /usr/local/elasticsearch

编辑配置文件

es1

cluster.name: myES
node.name: es1
node.master: true
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/log
bootstrap.mlockall: True
network.host: 10.80.2.83
http.port: 9200
http.enabled: true
transport.tcp.port: 9300
transport.tcp.compress: true
discovery.zen.ping.unicast.hosts: ["10.80.2.83:9300","10.80.2.84:9300"]
discovery.zen.minimum_master_nodes: 1
discovery.zen.ping.timeout: 3s
discovery.zen.ping.multicast.enabled: false
gateway.recover_after_nodes: 1
gateway.recover_after_time: 5m
gateway.expected_nodes: 2
cluster.routing.allocation.node_initial_primaries_recoveries: 4
indices.recovery.max_size_per_sec: 50mb

index :
  analysis :
    analyzer :
      default :
        tokenizer : keyword

config/elasticsearch.yml

#!/bin/sh

# check in case a user was using this mechanism
if [ "x$ES_CLASSPATH" != "x" ]; then
    cat >&2 << EOF
Error: Don\'t modify the classpath with ES_CLASSPATH. Best is to add
additional elements via the plugin mechanism, or if code must really be
added to the main classpath, add jars to lib/ (unsupported).
EOF
    exit 1
fi

ES_CLASSPATH="$ES_HOME/lib/elasticsearch-2.3.4.jar:$ES_HOME/lib/*"

if [ "x$ES_MIN_MEM" = "x" ]; then
    ES_MIN_MEM=8g
fi
if [ "x$ES_MAX_MEM" = "x" ]; then
    ES_MAX_MEM=8g
fi
if [ "x$ES_HEAP_SIZE" != "x" ]; then
    ES_MIN_MEM=$ES_HEAP_SIZE
    ES_MAX_MEM=$ES_HEAP_SIZE
fi

# min and max heap sizes should be set to the same value to avoid
# stop-the-world GC pauses during resize, and so that we can lock the
# heap in memory on startup to prevent any of it from being swapped
# out.
JAVA_OPTS="$JAVA_OPTS -Xms${ES_MIN_MEM}"
JAVA_OPTS="$JAVA_OPTS -Xmx${ES_MAX_MEM}"

# new generation
if [ "x$ES_HEAP_NEWSIZE" != "x" ]; then
    JAVA_OPTS="$JAVA_OPTS -Xmn${ES_HEAP_NEWSIZE}"
fi

# max direct memory
if [ "x$ES_DIRECT_SIZE" != "x" ]; then
    JAVA_OPTS="$JAVA_OPTS -XX:MaxDirectMemorySize=${ES_DIRECT_SIZE}"
fi

# set to headless, just in case
JAVA_OPTS="$JAVA_OPTS -Djava.awt.headless=true"

# Force the JVM to use IPv4 stack
if [ "x$ES_USE_IPV4" != "x" ]; then
  JAVA_OPTS="$JAVA_OPTS -Djava.net.preferIPv4Stack=true"
fi

# Add gc options. ES_GC_OPTS is unsupported, for internal testing
if [ "x$ES_GC_OPTS" = "x" ]; then
  ES_GC_OPTS="$ES_GC_OPTS -XX:+UseParNewGC"
  ES_GC_OPTS="$ES_GC_OPTS -XX:+UseConcMarkSweepGC"
  ES_GC_OPTS="$ES_GC_OPTS -XX:CMSInitiatingOccupancyFraction=75"
  ES_GC_OPTS="$ES_GC_OPTS -XX:+UseCMSInitiatingOccupancyOnly"
fi

JAVA_OPTS="$JAVA_OPTS $ES_GC_OPTS"

# GC logging options
if [ -n "$ES_GC_LOG_FILE" ]; then
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCDetails"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCTimeStamps"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCDateStamps"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintClassHistogram"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintTenuringDistribution"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCApplicationStoppedTime"
  JAVA_OPTS="$JAVA_OPTS -Xloggc:$ES_GC_LOG_FILE"

  # Ensure that the directory for the log file exists: the JVM will not create it.
  mkdir -p "`dirname \\"$ES_GC_LOG_FILE\\"`"
fi

# Causes the JVM to dump its heap on OutOfMemory.
JAVA_OPTS="$JAVA_OPTS -XX:+HeapDumpOnOutOfMemoryError"
# The path to the heap dump location, note directory must exists and have enough
# space for a full heap dump.
#JAVA_OPTS="$JAVA_OPTS -XX:HeapDumpPath=$ES_HOME/logs/heapdump.hprof"

# Disables explicit GC
JAVA_OPTS="$JAVA_OPTS -XX:+DisableExplicitGC"

# Ensure UTF-8 encoding by default (e.g. filenames)
JAVA_OPTS="$JAVA_OPTS -Dfile.encoding=UTF-8"

# Use our provided JNA always versus the system one
JAVA_OPTS="$JAVA_OPTS -Djna.nosys=true"

bin/elasticsearch.in.sh

es2

#集群名字
cluster.name: myES
#节点名字
node.name: es2
#是否有资格成为主节点
node.master: true
#数据目录
path.data: /data/elasticsearch/data
#日志目录
path.logs: /data/elasticsearch/log
#锁住内存
bootstrap.mlockall: True
#绑定的IP地址，可为0.0.0.0
#network.bind_host: 10.80.2.84
#该节点与其他节点交互的ip地址，
#network.publish_host: 10.80.2.84
#上面两个参数集合
network.host: 10.80.2.84
#对外服务端口
http.port: 9200
#是否使用http协议对外提供服务
http.enabled: true
#节点间交互tcp端口
transport.tcp.port: 9300
#传输数据时压缩
transport.tcp.compress: true
#集群中主节点初始化列表，通过这些节点自动发现加入其他节点
discovery.zen.ping.unicast.hosts: ["10.80.2.83:9300","10.80.2.84:9300"]
#保证集群中节点可以知道其他N各有主节点资格，默认1，es数超过2个，可设置大一些
discovery.zen.minimum_master_nodes: 1
#自动发现其他节点时ping超时时间，网络环境差，提高此参数
discovery.zen.ping.timeout: 3s
#是否打开多播发现节点，默认true
discovery.zen.ping.multicast.enabled: false
#集群中几个节点启动时进行数据恢复
gateway.recover_after_nodes: 1
#初始化数据恢复进程超时时间
gateway.recover_after_time: 5m
#这个es集群中节点数量，一旦这些数目节点启动，进行数据恢复
gateway.expected_nodes: 2
#初始化数据恢复时，并发恢复的线程个数
cluster.routing.allocation.node_initial_primaries_recoveries: 4
#数据恢复时最大带宽
indices.recovery.max_size_per_sec: 50mb
# 禁止分词，
index :
  analysis :
    analyzer :
      default :
        tokenizer : keyword

config/elasticsearch.yml

#!/bin/sh

# check in case a user was using this mechanism
if [ "x$ES_CLASSPATH" != "x" ]; then
    cat >&2 << EOF
Error: Don\'t modify the classpath with ES_CLASSPATH. Best is to add
additional elements via the plugin mechanism, or if code must really be
added to the main classpath, add jars to lib/ (unsupported).
EOF
    exit 1
fi

ES_CLASSPATH="$ES_HOME/lib/elasticsearch-2.3.4.jar:$ES_HOME/lib/*"

if [ "x$ES_MIN_MEM" = "x" ]; then
    ES_MIN_MEM=8g
fi
if [ "x$ES_MAX_MEM" = "x" ]; then
    ES_MAX_MEM=8g
fi
if [ "x$ES_HEAP_SIZE" != "x" ]; then
    ES_MIN_MEM=$ES_HEAP_SIZE
    ES_MAX_MEM=$ES_HEAP_SIZE
fi

# min and max heap sizes should be set to the same value to avoid
# stop-the-world GC pauses during resize, and so that we can lock the
# heap in memory on startup to prevent any of it from being swapped
# out.
JAVA_OPTS="$JAVA_OPTS -Xms${ES_MIN_MEM}"
JAVA_OPTS="$JAVA_OPTS -Xmx${ES_MAX_MEM}"

# new generation
if [ "x$ES_HEAP_NEWSIZE" != "x" ]; then
    JAVA_OPTS="$JAVA_OPTS -Xmn${ES_HEAP_NEWSIZE}"
fi

# max direct memory
if [ "x$ES_DIRECT_SIZE" != "x" ]; then
    JAVA_OPTS="$JAVA_OPTS -XX:MaxDirectMemorySize=${ES_DIRECT_SIZE}"
fi

# set to headless, just in case
JAVA_OPTS="$JAVA_OPTS -Djava.awt.headless=true"

# Force the JVM to use IPv4 stack
if [ "x$ES_USE_IPV4" != "x" ]; then
  JAVA_OPTS="$JAVA_OPTS -Djava.net.preferIPv4Stack=true"
fi

# Add gc options. ES_GC_OPTS is unsupported, for internal testing
if [ "x$ES_GC_OPTS" = "x" ]; then
  ES_GC_OPTS="$ES_GC_OPTS -XX:+UseParNewGC"
  ES_GC_OPTS="$ES_GC_OPTS -XX:+UseConcMarkSweepGC"
  ES_GC_OPTS="$ES_GC_OPTS -XX:CMSInitiatingOccupancyFraction=75"
  ES_GC_OPTS="$ES_GC_OPTS -XX:+UseCMSInitiatingOccupancyOnly"
fi

JAVA_OPTS="$JAVA_OPTS $ES_GC_OPTS"

# GC logging options
if [ -n "$ES_GC_LOG_FILE" ]; then
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCDetails"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCTimeStamps"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCDateStamps"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintClassHistogram"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintTenuringDistribution"
  JAVA_OPTS="$JAVA_OPTS -XX:+PrintGCApplicationStoppedTime"
  JAVA_OPTS="$JAVA_OPTS -Xloggc:$ES_GC_LOG_FILE"

  # Ensure that the directory for the log file exists: the JVM will not create it.
  mkdir -p "`dirname \\"$ES_GC_LOG_FILE\\"`"
fi

# Causes the JVM to dump its heap on OutOfMemory.
JAVA_OPTS="$JAVA_OPTS -XX:+HeapDumpOnOutOfMemoryError"
# The path to the heap dump location, note directory must exists and have enough
# space for a full heap dump.
#JAVA_OPTS="$JAVA_OPTS -XX:HeapDumpPath=$ES_HOME/logs/heapdump.hprof"

# Disables explicit GC
JAVA_OPTS="$JAVA_OPTS -XX:+DisableExplicitGC"

# Ensure UTF-8 encoding by default (e.g. filenames)
JAVA_OPTS="$JAVA_OPTS -Dfile.encoding=UTF-8"

# Use our provided JNA always versus the system one
JAVA_OPTS="$JAVA_OPTS -Djna.nosys=true"

bin/elasticsearch.in.sh