迁移场景 不同数据来源的集群，迁移方案会有差别，本章主要介以下场景的迁移方案。 源端为Elasticsearch的集群迁移 Elasticsearch集群的数据迁移有多种方式可以选择，例如使用Logstash、 CDM 、OBS备份与恢复、ESM、跨集群复制插件等进行数据迁移。 Logstash： CSS 提供Logstash，可以完成不同数据源和ES数据的迁移，还可以进行数据的清洗和加工。具体操作可以参考使用云服务Logstash全量迁移集群数据。 CDM：华为云服务提供的云迁移工具，实现不同云服务间的集群迁移能力。具体操作可以参考Elasticsearch整库迁移到 云搜索服务 。 备份与恢复：Elasticsearch提供备份恢复能力，可以把一个集群的数据备份到OBS，在另一个集群恢复数据，完成集群间的数据迁移。 CS S Elasticsearch集群之间的数据迁移操作可以参考使用备份与恢复迁移集群数据（源端为CSS Elasticsearch）；自建Elasticsearch集群或其他第三方Elasticsearch集群数据迁移到CSS Elasticsearch集群的操作可以参考使用备份与恢复迁移集群数据（源端为第三方Elasticsearch）。 源端为数据库的集群迁移

前提条件 目的端集群（Es-2）和源端集群（Es-1）处于可用状态。建议在业务空闲期进行集群迁移。 确认目的端集群（Es-2）和源端集群（Es-1）在同一个Region下。 确认目的端集群（Es-2）的版本大于等于源端集群（Es-1）的版本。 确认目的端集群（Es-2）的节点数大于源端集群（Es-1）的节点数的一半。 确认目的端集群（Es-2）的节点数大于等于源端集群（Es-1）的shard副本数。 确认目的端集群（Es-2）的CPU、MEM和Disk配置大于等于源端集群（Es-1）。

迁移时长 迁移过程的耗时长短依赖于源集群和目的集群的节点个数或索引shard个数。迁移过程分为备份阶段和恢复阶段，备份阶段耗时由源集群决定，恢复阶段耗时由目的集群决定。迁移总时长的评估公式如下： 当索引shard个数大于节点个数时 总时长(S)=（800G÷40MB÷源集群节点个数+800G÷40MB÷目的集群节点个数）x索引个数 当索引shard个数小于节点个数时 总时长(S)=（800G÷40MB÷源集群索引shard个数+800G÷40MB÷目的集群索引shard个数）x索引个数 评估公式是基于理想状态下（即单节点以最快速度40MB/s传输）的迁移时长，实际迁移时长还会受到网络、资源等因素影响。

集群节点数量 当CSS集群的构架与机型确定后，集群的节点数主要由业务对性能的要求决定。 表3 节点数量计算方式 类型 性能基线 节点数量计算方式 示例 写入节点 对于挂载云盘的节点，其单核写入性能基线为1MB/s。 对于超高IO型的节点，其单核写入性能基线为1.5MB/s。 写入节点数=业务峰值时的流量/单节点的核数/单核写入性能基线*副本数 业务峰值写入100MB/s，使用16u64g的节点，预计需要100/16/1*2 = 12个节点。 查询节点 相同节点，不同业务场景下的性能差异非常大，单节点的性能基线难以评估。这里以业务平均查询响应时间作为查询的性能基线进行测算。 查询节点数=QPS/{单节点的核数*3/2/平均查询响应时间(s)}*分片数量 查询QPS要求1000，平均查询响应时间100ms，索引规划3个分片，使用16u64g的节点，预计需要1000/{16*3/2/0.1}*3 = 12个节点。 节点数量 / 节点数量= 写入节点数 + 查询节点数 节点数= 写入节点数 + 查询节点数 = 24个节点数。 在同等集群性能的情况下，建议优先选择高配置少节点的集群。例如32C64G*3 节点的集群相比于8C16G*12节点的集群，在集群稳定性和扩容的便捷性上都有一定的优势。因为高配置的集群如果遇到性能瓶颈需要扩容，则只需要横向扩容，即向集群中加入更多同等配置的节点即可；而低配置的集群在扩容节点配置时，则需要纵向扩容。

集群版本 CSS中Elasticsearch搜索引擎的版本选择建议如下： 当首次使用CSS Elasticsearch集群，建议选择7.10.2或7.6.2版本。 如果是已有Elasticsearch集群迁移至CSS使用，当需要对迁移集群进行代码改造时，建议选择创建7.10.2或7.6.2版本的Elasticsearch集群；当希望基于之前版本完成集群迁移时，建议创建和原集群大版本一致的Elasticsearch集群。

集群架构 CSS支持读写分离、冷热分离、存算分离、角色分离、跨AZ部署等多种架构。 表1 集群架构适用的场景 架构 适用场景 用户价值 读写分离 生产业务，读多写少，数据写入后实时可见性要求低(10s+)。 高并发、低时延 冷热分离 日志业务，冷数据查询性能要求低。 低成本 存算分离 日志业务，冷数据不需要更新，并且冷数据查询性能要求低（10s+）；可以和冷热分离结合使用，构建“热-温-冷”3级存储。 低成本 角色分离 集群的规模较大、集群中的索引数量较多或集群可扩展性要求高。 高可用 跨AZ部署 对可用性要求非常高的生产业务，或采用本地盘时。 高可用

步骤四：全量迁移 使用putty登录准备工作中创建的Linux虚拟机。 执行python migrateTemplate.py 迁移索引模板。 执行 python migrateMapping.py迁移索引。 在Logstash集群管理页面，单击步骤一：创建Logstash集群中创建的集群名称，进入集群的基本信息页面。选择“配置中心”，或者直接单击目标集群操作列的“配置中心”，进入配置中心页面。 选择步骤三：配置Logstash全量迁移任务中所创建的配置文件，单击左上角的“启动”。 根据界面提示，选择是否启动Logstash服务会立刻开始迁移数据。 如果选择“是”，则可以在管道下面看到启动的配置文件。 数据迁移完毕检查数据一致性，使用putty登录linux虚拟机，执行python checkIndices.py 对比数据结果。

步骤一：创建Logstash集群 迁移数据使用Logstash，创建logstash服务需要费用，默认是按需收费，用户迁移完毕数据及时释放Logstash节省费用。 可以基于集群的索引不同创建多个Logstash集群分别配置不同的迁移任务。 登录 云搜索 服务管理控制台。 在“总览”或者“集群管理”页面，选择“Logstash”，进入Logsash类型集群管理页面。 单击“创建集群”，进入“创建集群”页面。 选择“当前区域”和“可用区”。 指定集群基本信息，选择“集群类型”和“集群版本”，并输入“集群名称”。 表3 基本参数说明 参数 说明 集群类型 选择“Logstash”。 集群版本 当前支持5.6.16、7.10.0。 对应ES集群是5.x, 6.x 选择logstash版本5.6.16， 对应ES版本是7.X 选择logstash版本7.10.0。 集群名称 自定义的集群名称，可输入的字符范围为4～32个字符，只能包含数字、字母、中划线和下划线，且必须以字母开头。 图2 基本信息配置 指定集群的主机规格相关参数。“节点数量”设置为“1”。“节点规格”选择“8U16G”，其余参数保持默认值。 图3 设置主机规格 设置集群的企业项目，保持默认值即可。 单击“下一步，网络配置”，设置集群的网络配置。 表4 参数说明 参数 说明 虚拟私有云 VPC即虚拟私有云，是通过逻辑方式进行网络隔离，提供安全、隔离的网络环境。 选择创建集群需要的VPC，单击“查看虚拟私有云”进入VPC服务查看已创建的VPC名称和ID。如果没有VPC，需要创建一个新的VPC。 说明： 此处您选择的VPC必须包含网段（CIDR），否则集群将无法创建成功。新建的VPC默认包含网段（CIDR）。 子网 通过子网提供与其他网络隔离的、可以独享的网络资源，以提高网络安全。 选择创建集群需要的子网，可进入VPC服务查看VPC下已创建的子网名称和ID。 安全组 安全组是一个逻辑上的分组，为同一个VPC内具有相同安全保护需求并相互信任的弹性云服务器提供访问策略。单击“查看安全组”可了解安全组详情。 说明： 请确保安全组的“端口范围/ICMP类型”为“Any”或者包含端口9200的端口范围。 图4 设置网络规格 单击“下一步：高级配置”，高级配置可选择默认配置和自定义。此样例保持默认配置即可。 单击“下一步：确认配置”，确认完成后单击“立即创建”开始创建集群。 单击“返回集群列表”，系统将跳转到“集群管理”页面。您创建的集群将展现在集群列表中，且集群状态为“创建中”，创建成功后集群状态会变为“可用”。

约束和限制 Logstash版本约束： CSS 支持5.5.1，6.3.2，6.5.4，7.1.1，7.6.2，7.10.2多个版本，迁移集群尽量保持大版本一致。 对应ES集群是5.x, 6.x 选择logstash版本5.6.16， 对应ES版本是7.X 选择logstash版本7.10.0。 集群迁移过程禁止修改索引，修改索引会导致原数据和目标数据内容不一致。 索引大小小于100G可以使用迁移任务不用单独分析索引，简化分析工作。

准备工作 创建迁移虚拟机。 创建迁移虚拟机，用于迁移源集群的元数据。 创建ECS虚拟机，虚拟机需要创建linux系统，规格选择2U4G。 测试虚拟机和源集群和目标集群保持连通性，执行命令curl http:// {ip}:{port}可以测试结果。 IP是源集群和目的集群访问地址，端口默认是9200，如果不是9200使用集群实际端口。 如下示例仅适用于非安全集群。 curl http://10.234.73.128:9200 { "name" : "voc_es_cluster_new-ess-esn-1-1", "cluster_name" : "voc_es_cluster_new", "cluster_uuid" : "1VbP7-39QNOx_R-llXKKtA", "version" : { "number" : "6.5.4", "build_flavor" : "default", "build_type" : "tar", "build_hash" : "d2ef93d", "build_date" : "2018-12-17T21:17:40.758843Z", "build_snapshot" : false, "lucene_version" : "7.5.0", "minimum_wire_compatibility_version" : "5.6.0", "minimum_index_compatibility_version" : "5.0.0" }, "Tagline" : "You Know, for Search" }

写入性能优化 基于Elasticsearch的数据写入流程分析，有以下几种性能优化方案。 表1 写入性能优化 序号 优化方案 方案说明 1 使用SSD盘或升级集群配置 使用SSD盘可以大幅提升数据写入与merge操作的速度，对应到CSS服务，建议选择“超高IO型”存储，或者超高IO型主机。 2 采用Bulk API 客户端采用批量数据的写入方式，每次批量写入的数据建议在1~10MB之间。 3 随机生成_id 如果采用指定_id的写入方式，数据写入时会先触发一次查询操作，进而影响数据写入性能。对于不需要通过_id检索数据的场景，建议使用随机生成的_id。 4 设置合适的分片数 分片数建议设置为集群数据节点的倍数，且分片的大小控制在50GB以内。 5 关闭副本 数据写入与查询错峰执行，在数据写入时关闭数据副本，待数据写入完成后再开启副本。 Elasticsearch 7.x版本中关闭副本的命令如下： PUT {index}/_settings { "number_of_replicas": 0 } 6 调整索引的刷新频率 数据批量写入时，可以将索引的刷新频率“refresh_interval”设置为更大的值或者设置为“-1”（表示不刷新），通过减少分片刷新次数提高写入性能。 Elasticsearch 7.x版本中，将更新时间设置为15s的命令如下： PUT {index}/_settings { "refresh_interval": "15s" } 7 优化写入线程数与写入队列大小 为应对突发流量，可以适当地提升写入线程数与写入队列的大小，防止突发流量导致出现错误状态码为429的情况。 Elasticsearch 7.x版本中，可以修改如下自定义参数实现写入优化：thread_pool.write.size，thread_pool.write.queue_size。 8 设置合适的字段类型 指定集群中各字段的类型，防止Elasticsearch默认将字段猜测为keyword和text的组合类型，增加不必要的数据量。其中keyword用于关键词搜索，text用于全文搜索。 对于不需要索引的字段，建议“index”设置为“false”。 Elasticsearch 7.x版本中，将字段“field1”设置为不建构索引的命令如下： PUT {index} { "mappings": { "properties": { "field1":{ "type": "text", "index": false } } } } 9 优化shard均衡策略 Elasticsearch默认采用基于磁盘容量大小的Load balance策略，在多节点场景下，尤其是在新扩容的节点上，可能出现shard在各节点上分配不均的问题。为避免这类问题，可以通过设置索引级别的参数“routing.allocation.total_shards_per_node”控制索引分片在各节点的分布情况。此参数可以在索引模板中配置，也可以修改已有索引的setting生效。 修改已有索引的setting的命令如下： PUT {index}/_settings { "index": { "routing.allocation.total_shards_per_node": 2 } }

数据写入流程 图1 数据写入流程 当从客户端往Elasticsearch中写入数据时，写入流程如下： 客户端向Node1发送写数据请求，此时Node1为协调节点。 节点Node1根据数据的_id将数据路由到分片2，此时请求会被转发到Node3，并执行写操作。 当主分片写入成功后，它将请求转发到Node2的副本分片上。当副本写入成功后，Node3将向协调节点报告写入成功，协调节点向客户端报告写入成功。 Elasticsearch中的单个索引由一个或多个分片(shard)组成，每个分片包含多个段（Segment），每一个Segment都是一个倒排索引。 图2 Elasticsearch的索引组成 将文档插入Elasticsearch时，文档首先会被写入缓冲区中，然后在刷新时定期从该缓冲区刷新到Segment中。刷新频率由refresh_interval参数控制，默认每1秒刷新一次。 图3 文档插入Elasticsearch的流程

方案架构 ELKB（Elasticsearch、Logstash、Kibana、Beats）提供了一整套日志场景解决方案，是目前主流的一种日志系统。其框架如图所示。 图1 统一日志管理平台框架 Beats是一个轻量级日志采集器，包括Filebeat、Metricbeat等。 Logstash用于对日志进行搜集与预处理，支持多种数据源与ETL处理方式。 Elasticsearch是个开源分布式搜索引擎，提供搜集、分析、存储数据等主要功能，CSS云搜索服务可以创建Elasticsearch集群。 Kibana是一个可视化工具，可以基于Kibana进行Web可视化查询，并制作BI报表。 本章以CSS、Filebeat、Logstash和Kibana为例，搭建一个统一日志管理平台。使用Filebeat采集ECS中的日志，发送到Logstash进行数据处理，再存储到CSS中，最后通过Kibana进行日志的可视化查询与分析。 ELKB系统中各组件的版本兼容性请参见 https://www.elastic.co/support/matrix#matrix_compatibility。

使用Spring Boot接入HTTPS集群（使用安全证书） 该场景适用于使用安全证书连接安全模式+HTTPS协议的集群。 获取安全证书（CloudSearchService.cer）。 登录云搜索服务控制台。 选择“集群管理”进入集群列表。 单击对应集群的名称，进入集群基本信息页面。 在“基本信息”页面，单击“HTTPS访问”后面的“下载证书”。 图1 下载证书 转换安全证书（CloudSearchService.cer）。将下载的安全证书上传到客户端机器上，使用keytool工具将“.cer”证书转换成Java可以读取的“.jks”证书格式。 在Linux系统中，执行如下命令转换证书。 keytool -import -alias newname -keystore ./truststore.jks -file ./CloudSearchService.cer 在Windows系统中，执行如下命令转换证书。 keytool -import -alias newname -keystore .\truststore.jks -file .\CloudSearchService.cer 其中，newname是由用户自定义的证书名称。 该命令执行后，会提示设置证书密码，并确认密码。请保存该密码，后续接入集群会使用。 application.properties配置文件： 1 2 3 elasticsearch.url=host1:9200,host2:9200 elasticsearch.username=username elasticsearch.password=password 表3 参数说明 参数 描述 host Elasticsearch集群节点的IP地址。 username 访问集群的用户名。 password 用户名对应的密码。 配置代码： com.xxx为项目目录，例如com.company.project。 com.xxx.repository为仓库目录，通过extends org.springframework.data.elasticsearch.repository.ElasticsearchRepository进行具体定义。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 package com.xxx.configuration; import org.elasticsearch.client.RestHighLevelClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.data.elasticsearch.client.ClientConfiguration; import org.springframework.data.elasticsearch.client.RestClients; import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration; import org.springframework.data.elasticsearch.repository.config.EnableElasticsearchRepositories; import java.io.File; import java.io.FileInputStream; import java.io.InputStream; import java.security.KeyStore; import java.security.SecureRandom; import java.security.cert.CertificateException; import java.security.cert.X509Certificate; import javax.net.ssl.HostnameVerifier; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLSession; import javax.net.ssl.TrustManager; import javax.net.ssl.TrustManagerFactory; import javax.net.ssl.X509TrustManager; @Configuration @EnableElasticsearchRepositories(basePackages = "com.xxx.repository") @ComponentScan(basePackages = "com.xxx") public class Config extends AbstractElasticsearchConfiguration { @Value("${elasticsearch.url}") public String elasticsearchUrl; @Value("${elasticsearch.username}") public String elasticsearchUsername; @Value("${elasticsearch.password}") public String elasticsearchPassword; @Override @Bean public RestHighLevelClient elasticsearchClient() { SSLContext sc = null; try { TrustManager[] tm = {new MyX509TrustManager(cerFilePath, cerPassword)}; sc = SSLContext.getInstance("SSL", "SunJSSE"); sc.init(null, tm, new SecureRandom()); } catch (Exception e) { e.printStackTrace(); } final ClientConfiguration clientConfiguration = ClientConfiguration.builder() .connectedTo(StringHostParse(elasticsearchUrl)) .usingSsl(sc, new NullHostNameVerifier()) .withBasicAuth(elasticsearchUsername, elasticsearchPassword) .build(); return RestClients.create(clientConfiguration).rest(); } private String[] StringHostParse(String hostAndPorts) { return hostAndPorts.split(","); } public static class MyX509TrustManager implements X509TrustManager { X509TrustManager sunJSSEX509TrustManager; MyX509TrustManager(String cerFilePath, String cerPassword) throws Exception { File file = new File(cerFilePath); if (!file.isFile()) { throw new Exception("Wrong Certification Path"); } System.out.println("Loading KeyStore " + file + "..."); InputStream in = new FileInputStream(file); KeyStore ks = KeyStore.getInstance("JKS"); ks.load(in, cerPassword.toCharArray()); TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509", "SunJSSE"); tmf.init(ks); TrustManager[] tms = tmf.getTrustManagers(); for (TrustManager tm : tms) { if (tm instanceof X509TrustManager) { sunJSSEX509TrustManager = (X509TrustManager) tm; return; } } throw new Exception("Couldn't initialize"); } @Override public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException { } @Override public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException { } @Override public X509Certificate[] getAcceptedIssuers() { return new X509Certificate[0]; } } public static class NullHostNameVerifier implements HostnameVerifier { @Override public boolean verify(String arg0, SSLSession arg1) { return true; } } } 其中，cerFilePath和cerPassword是生成的.jks证书的存放路径及其密码。

通过Spring Boot接入HTTP集群 该场景适用于连接非安全模式的集群或是安全模式+HTTP协议的集群。 配置文件： 1 2 3 4 elasticsearch.url=host1:9200,host2:9200 //非安全集群不用配置如下两行。 elasticsearch.username=username elasticsearch.password=password 表1 参数说明 参数 描述 host Elasticsearch集群节点的IP地址。 username 访问集群的用户名。 password 用户名对应的密码。 配置代码： com.xxx为项目目录，例如com.company.project。 com.xxx.repository为仓库目录，通过extends org.springframework.data.elasticsearch.repository.ElasticsearchRepository进行具体定义。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 package com.xxx.configuration; import org.elasticsearch.client.RestHighLevelClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.data.elasticsearch.client.ClientConfiguration; import org.springframework.data.elasticsearch.client.RestClients; import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration; import org.springframework.data.elasticsearch.repository.config.EnableElasticsearchRepositories; @Configuration @EnableElasticsearchRepositories(basePackages = "com.xxx.repository") @ComponentScan(basePackages = "com.xxx") public class Config extends AbstractElasticsearchConfiguration { @Value("${elasticsearch.url}") public String elasticsearchUrl; //非安全集群不用配置如下两个参数。 @Value("${elasticsearch.username}") public String elasticsearchUsername; @Value("${elasticsearch.password}") public String elasticsearchPassword; @Override @Bean public RestHighLevelClient elasticsearchClient() { final ClientConfiguration clientConfiguration = ClientConfiguration.builder() .connectedTo(StringHostParse(elasticsearchUrl)) //非安全集群无需配置withBasicAuth。 .withBasicAuth(elasticsearchUsername, elasticsearchPassword) .build(); return RestClients.create(clientConfiguration).rest(); } private String[] StringHostParse(String hostAndPorts) { return hostAndPorts.split(","); } }