电商网站建设 数商云,快速生成网页的软件,生产做网站表带的制造厂家,上海浦东做网站的公司文章目录 概览使用与ES交互索引创建索引查询索引删除文档创建修改文档局部修改文档查询文档删除全查询 整合SpringBootpom依赖application.ymlElasticsearchAutoConfigurationElasticsearchPropertiesElasticsearchConstantPersonSearchPageHelperPersonServiceBaseElasticsear… 文章目录 概览使用与ES交互索引创建索引查询索引删除文档创建修改文档局部修改文档查询文档删除全查询 整合SpringBootpom依赖application.ymlElasticsearchAutoConfigurationElasticsearchPropertiesElasticsearchConstantPersonSearchPageHelperPersonServiceBaseElasticsearchServicePersonServiceImplElasticsearchApplicationTests 集群架构搭建集群分布式架构原理故障转移分片控制写数据流程读数据流程搜索数据过程更新流程 原理倒排索引分析器文档搜索近实时搜索 优化合理设置分片数文件缓冲区数据预热冷热分离document模型设计 概览
Elasticsearch简称为 ES ES是一个开源的高扩展的分布式全文搜索引擎 是整个 ElasticStack 技术栈的核心。它可以近乎实时的存储、检索数据本身扩展性很好可以扩展到上百台服务器处理 PB 级别的数据。 Elastic Stack, 包括 Elasticsearch、 Kibana、 Beats 和 Logstash也称为 ELK Stack。能够安全可靠地获取任何来源、任何格式的数据然后实时地对数据进行搜索、分析和可视化。 Elasticsearch 是面向文档型数据库一条数据在这里就是一个文档。 Elasticsearch 里存储文档数据和关系型数据库 MySQL 存储数据的概念进行一个类比,如图: ES 里的 Index 可以看做一个库而 Types 相当于表 Documents 则相当于表的行。这里 Types 的概念已经被逐渐弱化 Elasticsearch 6.X 中一个 index 下已经只能包含一个type Elasticsearch 7.X 中, Type 的概念已经被删除了。
官网下载地址: https://www.elastic.co/cn/downloads/past-releases
使用
与ES交互
索引创建
发送put请求
http://127.0.0.1:9200/_indexname索引查询
发送get请求,查询单条
http://127.0.0.1:9200/_indexname查询所有索引,发送get请求
http://127.0.0.1:9200/_cat/indices索引删除
发送delete请求
http://127.0.0.1:9200/_indexname文档创建修改
发送post请求,第一次为创建,再一次发送为修改
http://127.0.0.1:9200/_indexname/_docname/idstr请求参数
{title:华为手机,category:小米,images:http://www.gulixueyuan.com/xm.jpg,price:3999.00
}文档局部修改
发送post请求
http://127.0.0.1:9200/_indexname/_update/idstr请求参数
{doc: {title:小米手机,category:小米}
}文档查询
发送get请求,查询单条
http://127.0.0.1:9200/_indexname/_docname/idstr文档删除
发送delete请求
http://127.0.0.1:9200/_indexname/_docname/idstr全查询
发送get请求,能看到全部数据
http://127.0.0.1:9200/_indexname/_searchurl带参查询,发get请求
http://127.0.0.1:9200/_indexname/_search?qcategory:小米请求体带参查询,发送get请求
http://127.0.0.1:9200/_indexname/_search查询category是华为和小米的,price大于2000,只显示title,显示第一页,每页显示两个,根据price降序
{query: {bool: {should: [{match: {category: 小米}},{match: {category: 华为}}]},filter: {range: {price: {gt: 2000}}}},_source: [title],from: 0,size: 2,sort: {price: {order: desc}}
}整合SpringBoot
pom依赖 dependencygroupIdorg.elasticsearch.client/groupIdartifactIdelasticsearch-rest-high-level-client/artifactIdversion7.5.2/version/dependencydependencygroupIdorg.elasticsearch.client/groupIdartifactIdelasticsearch-rest-client/artifactIdversion7.5.2/version/dependencydependencygroupIdorg.elasticsearch/groupIdartifactIdelasticsearch/artifactIdversion7.5.2/versionexclusionsexclusiongroupIdorg.elasticsearch.client/groupIdartifactIdelasticsearch-rest-client/artifactId/exclusionexclusiongroupIdorg.elasticsearch/groupIdartifactIdelasticsearch/artifactId/exclusion/exclusions/dependencyapplication.yml
demo:data:elasticsearch:cluster-name: elasticsearchcluster-nodes: [127.0.0.1:1001,127.0.0.1:1002,127.0.0.1:1003]index:number-of-replicas: 0number-of-shards: 3ElasticsearchAutoConfiguration
/*** ElasticsearchAutoConfiguration** since 2019-09-15 22:59*/
Configuration
RequiredArgsConstructor(onConstructor_ Autowired)
EnableConfigurationProperties(ElasticsearchProperties.class)
public class ElasticsearchAutoConfiguration {private final ElasticsearchProperties elasticsearchProperties;private ListHttpHost httpHosts new ArrayList();BeanConditionalOnMissingBeanpublic RestHighLevelClient restHighLevelClient() {ListString clusterNodes elasticsearchProperties.getClusterNodes();clusterNodes.forEach(node - {try {String[] parts StringUtils.split(node, :);Assert.notNull(parts, Must defined);Assert.state(parts.length 2, Must be defined as host:port);httpHosts.add(new HttpHost(parts[0], Integer.parseInt(parts[1]), elasticsearchProperties.getSchema()));} catch (Exception e) {throw new IllegalStateException(Invalid ES nodes property node , e);}});RestClientBuilder builder RestClient.builder(httpHosts.toArray(new HttpHost[0]));return getRestHighLevelClient(builder, elasticsearchProperties);}/*** get restHistLevelClient** param builder RestClientBuilder* param elasticsearchProperties elasticsearch default properties* return {link org.elasticsearch.client.RestHighLevelClient}* author fxbin*/private static RestHighLevelClient getRestHighLevelClient(RestClientBuilder builder, ElasticsearchProperties elasticsearchProperties) {// Callback used the default {link RequestConfig} being set to the {link CloseableHttpClient}builder.setRequestConfigCallback(requestConfigBuilder - {requestConfigBuilder.setConnectTimeout(elasticsearchProperties.getConnectTimeout());requestConfigBuilder.setSocketTimeout(elasticsearchProperties.getSocketTimeout());requestConfigBuilder.setConnectionRequestTimeout(elasticsearchProperties.getConnectionRequestTimeout());return requestConfigBuilder;});// Callback used to customize the {link CloseableHttpClient} instance used by a {link RestClient} instance.builder.setHttpClientConfigCallback(httpClientBuilder - {httpClientBuilder.setMaxConnTotal(elasticsearchProperties.getMaxConnectTotal());httpClientBuilder.setMaxConnPerRoute(elasticsearchProperties.getMaxConnectPerRoute());return httpClientBuilder;});// Callback used the basic credential authElasticsearchProperties.Account account elasticsearchProperties.getAccount();if (!StringUtils.isEmpty(account.getUsername()) !StringUtils.isEmpty(account.getUsername())) {final CredentialsProvider credentialsProvider new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(account.getUsername(), account.getPassword()));}return new RestHighLevelClient(builder);}}ElasticsearchProperties
/*** ElasticsearchProperties** version v1.0* since 2019-09-15 22:58*/
Data
Builder
Component
NoArgsConstructor
AllArgsConstructor
ConfigurationProperties(prefix demo.data.elasticsearch)
public class ElasticsearchProperties {/*** 请求协议*/private String schema http;/*** 集群名称*/private String clusterName elasticsearch;/*** 集群节点*/NotNull(message 集群节点不允许为空)private ListString clusterNodes new ArrayList();/*** 连接超时时间(毫秒)*/private Integer connectTimeout 1000;/*** socket 超时时间*/private Integer socketTimeout 30000;/*** 连接请求超时时间*/private Integer connectionRequestTimeout 500;/*** 每个路由的最大连接数量*/private Integer maxConnectPerRoute 10;/*** 最大连接总数量*/private Integer maxConnectTotal 30;/*** 索引配置信息*/private Index index new Index();/*** 认证账户*/private Account account new Account();/*** 索引配置信息*/Datapublic static class Index {/*** 分片数量*/private Integer numberOfShards 3;/*** 副本数量*/private Integer numberOfReplicas 2;}/*** 认证账户*/Datapublic static class Account {/*** 认证用户*/private String username;/*** 认证密码*/private String password;}}ElasticsearchConstant
/*** ElasticsearchConstant** version v1.0* since 2019-09-15 23:03*/
public interface ElasticsearchConstant {/*** 索引名称*/String INDEX_NAME person;/*** 文档名称(字段名称)*/String COLUMN_NAME_1 column_1;String COLUMN_NAME_2 column_2;String COLUMN_NAME_3 column_3;String COLUMN_NAME_4 column_4;String COLUMN_NAME_5 column_5;/*** 高亮标签*/String TAG_HIGH_LIGHT_START label stylecolor:red;String TAG_HIGH_LIGHT_END /label;}Person
/*** Person** version v1.0* since 2019-09-15 23:04*/
Data
Builder
NoArgsConstructor
AllArgsConstructor
public class Person implements Serializable {private static final long serialVersionUID 8510634155374943623L;/*** 主键*/private Long id;/*** 名字*/private String name;/*** 国家*/private String country;/*** 年龄*/private Integer age;/*** 生日*/private Date birthday;/*** 介绍*/private String remark;}SearchPageHelper
/*** author: whitepure* date: 2023/1/6 11:25* description: SearchPageHelper*/
Data
Accessors(chain true)
public class SearchPageHelperE {private Long current;private Long pageSize;private Long total;private ListE records;}PersonService
/*** PersonService** version v1.0* since 2019-09-15 23:07*/
public interface PersonService {/*** create Index** param index elasticsearch index name* author fxbin*/void createIndex(String index);/*** delete Index** param index elasticsearch index name* author fxbin*/void deleteIndex(String index);/*** insert document source** param index elasticsearch index name* param list data source* author fxbin*/void insert(String index, ListPerson list);/*** update document source** param index elasticsearch index name* param list data source* author fxbin*/void update(String index, ListPerson list);/*** delete document source** param person delete data source and allow null object* author fxbin*/void delete(String index, Nullable Person person);/*** search all doc records** param index elasticsearch index name* return person list* author fxbin*/ListPerson searchList(String index);/*** 分页查询** param searchRequest search condition* return search list*/SearchPageHelperPerson searchPage(SearchRequest searchRequest);}BaseElasticsearchService
/*** BaseElasticsearchService** version 1.0v* since 2019-09-16 15:44*/
Slf4j
public abstract class BaseElasticsearchService {Resourceprotected RestHighLevelClient client;Resourceprivate ElasticsearchProperties elasticsearchProperties;protected static final RequestOptions COMMON_OPTIONS;static {RequestOptions.Builder builder RequestOptions.DEFAULT.toBuilder();// 默认缓冲限制为100MB此处修改为30MB。builder.setHttpAsyncResponseConsumerFactory(new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));COMMON_OPTIONS builder.build();}/*** create elasticsearch index (asyc)** param index elasticsearch index* author fxbin*/protected void createIndexRequest(String index) {try {CreateIndexRequest request new CreateIndexRequest(index);// Settings for this indexrequest.settings(Settings.builder().put(index.number_of_shards, elasticsearchProperties.getIndex().getNumberOfShards()).put(index.number_of_replicas, elasticsearchProperties.getIndex().getNumberOfReplicas()));CreateIndexResponse createIndexResponse client.indices().create(request, COMMON_OPTIONS);log.info( whether all of the nodes have acknowledged the request : {}, createIndexResponse.isAcknowledged());log.info( Indicates whether the requisite number of shard copies were started for each shard in the index before timing out :{}, createIndexResponse.isShardsAcknowledged());} catch (IOException e) {throw new ElasticsearchException(创建索引 { index } 失败);}}/*** delete elasticsearch index** param index elasticsearch index name* author fxbin*/protected void deleteIndexRequest(String index) {DeleteIndexRequest deleteIndexRequest buildDeleteIndexRequest(index);try {client.indices().delete(deleteIndexRequest, COMMON_OPTIONS);} catch (IOException e) {throw new ElasticsearchException(删除索引 { index } 失败);}}/*** build DeleteIndexRequest** param index elasticsearch index name* author fxbin*/private static DeleteIndexRequest buildDeleteIndexRequest(String index) {return new DeleteIndexRequest(index);}/*** build IndexRequest** param index elasticsearch index name* param id request object id* param object request object* return {link org.elasticsearch.action.index.IndexRequest}* author fxbin*/protected static IndexRequest buildIndexRequest(String index, String id, Object object) {return new IndexRequest(index).id(id).source(BeanUtil.beanToMap(object), XContentType.JSON);}/*** exec updateRequest** param index elasticsearch index name* param id Document id* param object request object* author fxbin*/protected void updateRequest(String index, String id, Object object) {try {UpdateRequest updateRequest new UpdateRequest(index, id).doc(BeanUtil.beanToMap(object), XContentType.JSON);client.update(updateRequest, COMMON_OPTIONS);} catch (IOException e) {throw new ElasticsearchException(更新索引 { index } 数据 { object } 失败);}}/*** exec deleteRequest** param index elasticsearch index name* param id Document id* author fxbin*/protected void deleteRequest(String index, String id) {try {DeleteRequest deleteRequest new DeleteRequest(index, id);client.delete(deleteRequest, COMMON_OPTIONS);} catch (IOException e) {throw new ElasticsearchException(删除索引 { index } 数据id { id } 失败);}}/*** 查询全部** param indices elasticsearch 索引名称* return {link SearchResponse}*/protected SearchResponse search(String... indices) {SearchRequest searchRequest new SearchRequest(indices);SearchSourceBuilder searchSourceBuilder new SearchSourceBuilder();searchSourceBuilder.query(QueryBuilders.matchAllQuery());searchRequest.source(searchSourceBuilder);return search(searchRequest);}/*** 查询全部** param searchRequest 查询条件* return {link SearchResponse}*/protected SearchResponse search(SearchRequest searchRequest) {SearchResponse searchResponse;try {searchResponse client.search(searchRequest, COMMON_OPTIONS);} catch (IOException e) {throw new org.elasticsearch.ElasticsearchException(查询索引 %s 失败 , e, Arrays.toString(Arrays.stream(searchRequest.indices()).toArray()));}return searchResponse;}
}PersonServiceImpl
/*** PersonServiceImpl** version v1.0* since 2019-09-15 23:08*/
Service
public class PersonServiceImpl extends BaseElasticsearchService implements PersonService {Overridepublic void createIndex(String index) {createIndexRequest(index);}Overridepublic void deleteIndex(String index) {deleteIndexRequest(index);}SneakyThrowsOverridepublic void insert(String index, ListPerson list) {for (Person person : list) {IndexRequest request buildIndexRequest(index, String.valueOf(person.getId()), person);client.index(request, COMMON_OPTIONS);}}Overridepublic void update(String index, ListPerson list) {list.forEach(person - updateRequest(index, String.valueOf(person.getId()), person));}Overridepublic void delete(String index, Person person) {if (ObjectUtils.isEmpty(person)) {// 如果person 对象为空则删除全量searchList(index).forEach(p - {deleteRequest(index, String.valueOf(p.getId()));});}deleteRequest(index, String.valueOf(person.getId()));}Overridepublic ListPerson searchList(String index) {return toSearchList(search(index));}Overridepublic SearchPageHelperPerson searchPage(SearchRequest searchRequest) {SearchResponse searchResponse search(searchRequest);TotalHits totalHits searchResponse.getHits().getTotalHits();return new SearchPageHelperPerson().setTotal(totalHits null ? 0 : totalHits.value).setRecords(toSearchList(searchResponse)).setPageSize(20L);}private ListPerson toSearchList(SearchResponse searchResponse) {SearchHit[] hits searchResponse.getHits().getHits();ListPerson searchList new ArrayList();Arrays.stream(hits).forEach(hit - {MapString, Object sourceAsMap hit.getSourceAsMap();// 处理高亮数据MapString, HighlightField highlightFields hit.getHighlightFields();highlightFields.forEach((k, v) - {if (v ! null v.getFragments().length 0) {sourceAsMap.put(k, StrUtil.strip(Arrays.toString(v.getFragments()), []));}});Person search BeanUtil.mapToBean(sourceAsMap, Person.class, true);searchList.add(search);});return searchList;}}ElasticsearchApplicationTests
RunWith(SpringRunner.class)
SpringBootTest
public class ElasticsearchApplicationTests {Autowiredprivate PersonService personService;/*** 测试删除索引*/Testpublic void deleteIndexTest() {personService.deleteIndex(ElasticsearchConstant.INDEX_NAME);}/*** 测试创建索引*/Testpublic void createIndexTest() {personService.createIndex(ElasticsearchConstant.INDEX_NAME);}/*** 测试新增*/Testpublic void insertTest() {ListPerson list new ArrayList();list.add(Person.builder().age(11).birthday(new Date()).country(CN).id(1L).name(哈哈).remark(test1).build());list.add(Person.builder().age(22).birthday(new Date()).country(US).id(2L).name(hiahia).remark(test2).build());list.add(Person.builder().age(33).birthday(new Date()).country(ID).id(3L).name(呵呵).remark(test3).build());personService.insert(ElasticsearchConstant.INDEX_NAME, list);}/*** 测试更新*/Testpublic void updateTest() {Person person Person.builder().age(33).birthday(new Date()).country(ID_update).id(3L).name(呵呵update).remark(test3_update).build();ListPerson list new ArrayList();list.add(person);personService.update(ElasticsearchConstant.INDEX_NAME, list);}/*** 测试删除*/Testpublic void deleteTest() {personService.delete(ElasticsearchConstant.INDEX_NAME, Person.builder().id(1L).build());}/*** 测试查询*/Testpublic void searchListTest() {ListPerson personList personService.searchList(ElasticsearchConstant.INDEX_NAME);System.out.println(personList);}/*** 测试分页查询*/Testpublic void searchPageTest(){int current 1;int pageSize 20;int maxCurrent 500;// 此处最大页数设置为500 为es浅分页; 如需深度分页需更换分页方法并移除此条件if (current maxCurrent) {return;}// 构造查询条件SearchRequest searchRequest new SearchRequest(ElasticsearchConstant.INDEX_NAME);searchRequest.source(new SearchSourceBuilder().trackTotalHits(true)// 查询条件.query(QueryBuilders.boolQuery()// and.must(QueryBuilders.termQuery(ElasticsearchConstant.COLUMN_NAME_2, true))// or.must(QueryBuilders.boolQuery().should(QueryBuilders.matchQuery(ElasticsearchConstant.COLUMN_NAME_1, 1)).should(QueryBuilders.matchQuery(ElasticsearchConstant.COLUMN_NAME_2, 2)).should(QueryBuilders.matchQuery(ElasticsearchConstant.COLUMN_NAME_3, 3))))// 分页.from((current - 1) * pageSize).size(pageSize)// 相关度排序: SortBuilders.scoreSort().sort(// 字段排序 可根据时间SortBuilders.fieldSort(ElasticsearchConstant.COLUMN_NAME_2).order(SortOrder.DESC))// 高亮字段.highlighter(new HighlightBuilder().requireFieldMatch(true).preTags(ElasticsearchConstant.TAG_HIGH_LIGHT_START).field(ElasticsearchConstant.COLUMN_NAME_1).field(ElasticsearchConstant.COLUMN_NAME_2).field(ElasticsearchConstant.COLUMN_NAME_3).postTags(ElasticsearchConstant.TAG_HIGH_LIGHT_END)));SearchPageHelperPerson personSearchPageHelper personService.searchPage(searchRequest);System.out.println(personSearchPageHelper);}}集群架构
一个运行中的 Elasticsearch 实例称为一个节点而集群是由一个或者多个拥有相同 cluster.name 配置的节点组成 它们共同承担数据和负载的压力。 当有节点加入集群中或者从集群中移除节点时集群将会重新平均分布所有的数据。
当一个节点被选举成为主节点时 它将负责管理集群范围内的所有变更例如增加、 删除索引或者增加、删除节点等。 而主节点并不需要涉及到文档级别的变更和搜索等操作所以当集群只拥有一个主节点的情况下即使流量的增加它也不会成为瓶颈。 任何节点都可以成为主节点。我们的示例集群就只有一个节点所以它同时也成为了主节点。
作为用户我们可以将请求发送到集群中的任何节点 包括主节点。 每个节点都知道任意文档所处的位置并且能够将我们的请求直接转发到存储我们所需文档的节点。 无论我们将请求发送到哪个节点它都能负责从各个包含我们所需文档的节点收集回数据并将最终结果返回给客户端。
搭建集群
创建 elasticsearch-cluster 文件夹在内部复制三个 elasticsearch 服务。 修改节点配置; config/elasticsearch.yml 文件 node-1001 节点#集群名称节点之间要保持一致
cluster.name: my-elasticsearch
#节点名称集群内要唯一
node.name: node-1001
node.master: true
node.data: true
#ip 地址
network.host: localhost
#http 端口
http.port: 1001
#tcp 监听端口
transport.tcp.port: 9301
#discovery.seed_hosts: [localhost:9301, localhost:9302,localhost:9303]
#discovery.zen.fd.ping_timeout: 1m
#discovery.zen.fd.ping_retries: 5
#集群内的可以被选为主节点的节点列表
#cluster.initial_master_nodes: [node-1, node-2,node-3]
#跨域配置
#action.destructive_requires_name: true
http.cors.enabled: true
http.cors.allow-origin: *node-1002 节点 #集群名称节点之间要保持一致cluster.name: my-elasticsearch#节点名称集群内要唯一node.name: node-1002node.master: truenode.data: true#ip 地址network.host: localhost#http 端口http.port: 1002#tcp 监听端口transport.tcp.port: 9302discovery.seed_hosts: [localhost:9301]discovery.zen.fd.ping_timeout: 1mdiscovery.zen.fd.ping_retries: 5#集群内的可以被选为主节点的节点列表#cluster.initial_master_nodes: [node-1, node-2,node-3]#跨域配置#action.destructive_requires_name: truehttp.cors.enabled: truehttp.cors.allow-origin: *node-1003 节点#集群名称节点之间要保持一致
cluster.name: my-elasticsearch
#节点名称集群内要唯一
node.name: node-1003
node.master: true
node.data: true
#ip 地址
network.host: localhost
#http 端口
http.port: 1003
#tcp 监听端口
transport.tcp.port: 9303
#候选主节点的地址在开启服务后可以被选为主节点
discovery.seed_hosts: [localhost:9301, localhost:9302]
discovery.zen.fd.ping_timeout: 1m
discovery.zen.fd.ping_retries: 5
#集群内的可以被选为主节点的节点列表
#cluster.initial_master_nodes: [node-1, node-2,node-3]
#跨域配置
#action.destructive_requires_name: true
http.cors.enabled: true
http.cors.allow-origin: *启动集群; 点击 bin\elasticsearch.bat
如果启动不起来可能原因是分配内存不足需要修改 config\jvm.options 文件中的内存属性
启动之后使用ES可视化工具查看,可使用elasticsearch-head,ElasticHD
分布式架构原理
ElasticSearch 设计的理念就是分布式搜索引擎底层其实还是基于 lucene 的。核心思想就是在多台机器上启动多个 ES 进程实例组成了一个 ES 集群。
ES分布式架构实际上就是对index的拆分,将index拆分成多个分片(shard),将分片分别放到不同的ES上实现集群部署. 分片优点:
支持横向扩展: 比如你数据量是 3T3 个 shard每个 shard 就 1T 的数据若现在数据量增加到 4T,怎么扩展很简单重新建一个有 4 个 shard 的索引将数据导进去;提高性能: 数据分布在多个 shard即多台服务器上所有的操作都会在多台机器上并行分布式执行提高了吞吐量和性能;
分片的数据实际上是有多个备份存在的,会存在一个主分片,还有几个副本分片. 当写入数据的时候先写入主分片,然后并行将数据同步到副本分片上;当读数据的时候会获取到所有分片,负载均衡轮询读取.
当某个节点宕机了,还有其他分片副本保存在其他的机器上,从而实现了高可用. 如果是非主节点宕机了那么会由主节点让那个宕机节点上的主分片的数据转移到其他机器上的副本数据。接着你要是修复了那个宕机机器重启了之后主节点会控制将缺失的副本数据分配过去同步后续修改的数据之类的让集群恢复正常. 如果是主节点宕机,那么会重新选举一个节点为主节点.
故障转移
在一个网络环境里失败随时都可能发生在某个分片/节点不知怎么的就处于离线状态或者由于任何原因消失了这种情况下有一个故障转移机制是非常有用并且是强烈推荐的。为此目的Elasticsearch 允许你创建分片的一份或多份拷贝这些拷贝叫做复制分片。
当集群中只有一个节点在运行时意味着会有一个单点故障问题——没有冗余。 幸运的是我们只需再启动一个节点即可防止数据丢失。当你在同一台机器上启动了第二个节点时只要它和第一个节点有同样的 cluster.name 配置它就会自动发现集群并加入到其中。
ES最好部署3个以上的节点并且配置仲裁数大于一半节点防止master选举的脑裂问题。
当一个节点掉线如果该节点是master节点则通过比较node ID选择较小ID的节点为master然后由master节点决定分片如何重新分配。同理新加入节点也是由master决定如何分配分片 关于master的选举: 主要是由ZenDiscovery模块负责,包含Ping(节点之间通过这个RPC来发现彼此)和Unicast单播模块包含-一个主机列表以控制哪些节点需要ping通这两部分. 首先对所有可以成为master的节点(可以配置)根据nodeId排序,每次选举每个节点都把自己所知道节点排一次序,然后选出第一个,暂且认为它是master节点 如果对某个节点的投票数达到一定的值可以成为master节点数n/21并且该节点自己也选举自己,那这个节点就是master。否则重新选举一直到满足上述条件 ES在主节点上产生分歧产生多个主节点从而使集群分裂使得集群处于异常状态。这个现象叫做脑裂。脑裂问题其实就是同一个集群的不同节点对于整个集群的状态有不同的理解导致操作错乱类似于精神分裂。 分片控制
当写入一个文档的时候文档会被存储到一个主分片中。 Elasticsearch 集群如何知道一个文档应该存放到哪个分片中呢
Elasticsearch 集群路由计算公式:
shard hash(routing) % number_of_primary_shardsrouting 是一个可变值默认是文档的 _id 也可以设置成一个自定义的值。 routing 通过hash 函数生成一个数字然后这个数字再除以 number_of_primary_shards 主分片的数量后得到余数 。这个分布在 0 到 number_of_primary_shards-1 之间的余数就是我们所寻求的文档所在分片的位置。
这也就是创建索引的时候主分片的数量永远也不会改变的原因,如果数量变化了,那么所有之前路由的值都会无效,文档也再也找不到了.
用户可以访问任何一个节点获取数据,因为存放的规则一致(副本和主分片存放的数据一致),这个节点称之为协调节点.如果当前节点访问量较大可能被转到其他节点上,所以当发送请求的时候,为了扩展负载,更好的做法是轮询集群中所有的节点.
写数据流程 客户端请请求任意集群节点(协调节点);协调节点将请求转换到指定节点(路由计算);主分片需要将数据保存;主分片将保存数据的请求发送到各个副本;各个副本保存后,进行响应;主分片进行响应;客户端获取响应;
在客户端收到成功响应时文档变更已经在主分片和所有副本分片执行完成变更是安全的。有一些可选的请求参数允许您影响这个过程可能以数据安全为代价提升性能。
设置 consistency 参数值会影响写入操作.consistency 参数的值可以设为:
one 只要主分片状态 ok 就允许执行写操作;all必须要主分片和所有副本分片的状态没问题才允许执行写操作;quorum默认值为quorum , 即大多数的分片副本状态没问题就允许执行写操作;
当consistency值设置为quorum时,如果没有足够的副本分片Elasticsearch 会等待.默认情况下,它最多等待 1 分钟,可以使用timeout参数使它更早终止.
读数据流程
客户端发送查询请求到协调节点;协调节点计算数据所在的分片及全部的副本位置,为了能负载均衡要轮询所有的分片;将请求转发给具体的节点;节点返回查询结果,将结果返回给客户端;
在处理读取请求时协调结点在每次请求的时候都会通过轮询所有的副本分片来达到负载均衡。在文档被检索时已经保存的数据可能已经存在于主分片上但是还没有复制到副本分片。 在这种情况下副本分片可能会报告文档不存在但是主分片可能成功返回文档。 一旦索引请求成功返回给用户文档在主分片和副本分片都是可用的。
搜索数据过程
客户端发送请求到一个协调节点;协调节点计算数据所在的分片及全部的副本位置;每个分片将自己的搜索结果其实就是一些 doc id 返回给协调节点由协调节点进行数据的合并、排序、分页等操作产出最终结果;接着由协调节点根据 doc id 去各个节点上拉取实际的 document 数据最终返回给客户端;
更新流程
客户端向某一节点发送更新请求;将请求转发到主分片所在的节点;从主分片检索文档修改_source字段中的JSON并且尝试重新索引主分片的文档。如果文档已经被另一个进程修改,它会重试步骤3 ,超过retry_on_conflict次后放弃;如果主节点成功地更新文档它将新版本的文档并行转发到副本分片重新建立索引。一旦所有副本分片都返回成功主节点向协调节点也返回成功协调节点向客户端返回成功;
在步骤4中,主分片把更改转发到副本分片时 它不会转发更新请求。 相反它转发完整文档的新版本。请记住这些更改将会异步转发到副本分片并且不能保证它们以发送它们相同的顺序到达。 如果 Elasticsearch 仅转发更改请求则可能以错误的顺序应用更改导致得到损坏的文档。
原理
分片是Elasticsearch最小的工作单元。传统的数据库每个字段存储单个值但这对全文检索并不够。文本字段中的每个单词需要被搜索对数据库意味着需要单个字段有索引多值的能力。最好的支持是一个字段多个值需求的数据结构是倒排索引。
倒排索引
Elasticsearch 使用一种称为 倒排索引 的结构它适用于快速的全文搜索。一个倒排索引由文档中所有不重复词的列表构成对于其中每个词有一个包含它的文档列表。 倒排索引Inverted Index也叫反向索引有反向索引必有正向索引。通俗地来讲正向索引是通过key找value反向索引则是通过value找key。
倒排索引示例:
| value | key |
|----------|-------------------|
| my name is zhangsan | 1001 | | key | value|
|---------|------|
| name | 1001 |
| zhang | 1001 |
| zhangsan| 1001 |倒排索引搜索过程: 查询单词是否在词典中,如果不在搜索结束,如果在词典中需要查询单词在倒排列表中的指针,获取单词对应的文档ID,根据文档ID查询时哪一条数据
词条: 索引中最小的存储和查询单元 词典: 词条的集合;一般用hash表或Btree存储 倒排表: 记录了出现过某个单词的所有文档的文档列表及单词在该文档中出现的位置信息,每条记录称为一个倒排项.根据倒排列表,即可获知哪些文档包含某个单词.
分析器
倒排索引总是和分词分不开的,中文分词和英文分词是不一样的,所以就需要分析器.
分析器的主要功能是将一块文本分成适合于倒排索引的独立词条,分析器组成:
字符过滤器: 在分词前整理字符串,一个字符过滤器可以用来去掉 HTML,或者将 转化成 and;分词器: 字符串被分词器分为单个的词条,一个简单的分词器遇到空格和标点的时候,可能会将文本拆分成词条;词单元过滤器: 按顺序通过每个过滤器,这个过程可能会改变词条例如小写化Quick 删除词条例如 像 a and the 等无用词或者增加词条例如像jump和leap这种同义词;
Elasticsearch附带了可以直接使用的预包装的分析器:
标准分析器: 默认使用的分析器。它是分析各种语言文本最常用的选择。它根据Unicode 联盟定义的单词边界划分文本。删除绝大部分标点;简单分析器: 在任何不是字母的地方分隔文本将词条小写;空格分析器: 在空格的地方划分文本;语言分析器: 考虑指定语言的特点,根据语法进行分词; 例如英语分析器附带了一组英语无用词常用单词例如and或者the ,它们对相关性没有多少影响它们会被删除;
常用中文分词器: ik分词器, 将解压后的后的文件夹放入 ES 根目录下的 plugins 目录下重启 ES 即可使用.
文档搜索
早期的全文检索会为整个文档集合建立一个很大的倒排索引并将其写入到磁盘。 被写入的索引不可变化,一旦新的索引就绪旧的就会被其替换.如果你需要让一个新的文档可被搜索你需要重建整个索引。这要么对一个索引所能包含的数据量造成了很大的限制要么对索引可被更新的频率造成了很大的限制。
如何在保留不变性的前提下实现倒排索引的更新?
用更多的索引。通过增加新的补充索引来反映新的修改而不是直接重写整个倒排索引。每一个倒排索引都会被轮流查询到,从最早的开始查询完后再对结果进行合并。
当一个文档被删除时它实际上只是在文件中被标记删除。一个被标记删除的文档仍然可以被查询匹配到但它会在最终结果被返回前从结果集中过滤掉。 文档更新也是类似的操作方式:当一个文档被更新时旧版本文档被标记删除文档的新版本被索引到一个新的段中。可能两个版本的文档都会被一个查询匹配到但被删除的那个旧版本文档在结果集返回前就已经被移除。 当一个查询被触发所有已知的段按顺序被查询。词项统计会对所有段的结果进行聚合,此时会将标记删除的数据真正的删除.
近实时搜索
Elasticsearch 的主要功能就是搜索,但是Elasticsearch的搜索功能不是实时的,而是近实时的,主要原因在于ES搜索是分段搜索.
ES中的每一段就是一个倒排索引,最新的数据更新会体现在最新的段中,而最新的段落盘之后ES才能进行搜索,所以磁盘性能极大影响了ES软件的搜索.ES的主要作用就是快速准确的获取想要的数据,所以降低处理数据的延迟就显得尤为重要.
ES近实时搜索实现: 一个文档被索引之后就会被添加到内存缓冲区并且追加到了 translog 事务日志中;(先写入索引中,再写入到日志中,目的防止数据丢失,类似数据库中的事务日志)将内存缓冲区中的分片刷新到磁盘中(refresh);此时缓冲区的数据可被搜索,当完全将数据写入磁盘会清空缓冲区中的数据;随着不断的刷写,磁盘中的文件会越来越多,此时需要文件段合并;当一个新的索引文件产生之后,文件的更新,删除便会体现出,此时在合并文件的时候便会真正的将数据删除;小的段被合并到大的段然后这些大的段再被合并到更大的段;
优化
Elasticsearch 在数据量很大的情况下数十亿级别如何提高查询效率
合理设置分片数
分片和副本的设计为 ES 提供了支持分布式和故障转移的特性但并不意味着分片和副本是可以无限分配的。而且索引的分片完成分配后由于索引的路由机制我们是不能重新修改分片数的.否则将无法找到对应的数据.
控制每个分片占用的硬盘容量不超过 ES 的最大 JVM 的堆空间设置因此如果索引的总容量在 500G 左右那分片大小在 16 个左右即可;考虑一下 node 数量一般一个节点有时候就是一台物理机如果分片数过多大大超过了节点数很可能会导致一个节点上存在多个分片一旦该节点故障即使保持了 1 个以上的副本同样有可能会导致数据丢失集群无法恢复。所以 一般都设置分片数不超过节点数的 3 倍;主分片副本和节点最大数之间数量我们分配的时候可以参考 节点数主分片数 *副本数1;
文件缓冲区
往 ES 里写的数据实际上都写到磁盘文件里去了查询的时候操作系统会将磁盘文件里的数据自动缓存到 filesystem cache 里面去;ES 的搜索引擎严重依赖于底层的 filesystem cache 你如果给 filesystem cache 更多的内存尽量让内存可以容纳所有的 idx segment file 索引数据文件那么你搜索的时候就基本都是走内存的性能会非常高。 案例: 某个公司 ES 节点有 3 台机器每台机器看起来内存很多64G总内存就是 64 * 3 192G 。每台机器给 ES jvm heap 是 32G 那么剩下来留给 filesystem cache 的就是每台机器才 32G 总共集群里给 filesystem cache 的就是 32 * 3 96G 内存。而此时整个磁盘上索引数据文件在 3 台机器上一共占用了 1T 的磁盘容量ES 数据量是 1T 那么每台机器的数据量是 300G 。这样性能好吗 filesystem cache 的内存才 100G十分之一的数据可以放内存其他的都在磁盘然后你执行搜索操作大部分操作都是走磁盘性能肯定差。 归根结底你要让 ES 性能要好最佳的情况下就是你的机器的内存至少可以容纳你的总数据量的一半。
根据生产环境实践经验最佳的情况下是仅仅在 ES 中就存少量的数据就是你要用来搜索的那些索引如果内存留给 filesystem cache 的是 100G那么你就将索引数据控制在 100G 以内这样的话你的数据几乎全部走内存来搜索性能非常之高一般可以在 1 秒以内。
比如说你现在有一行数据。 id,name,age … 30 个字段。但是你现在搜索只需要根据 id,name,age 三个字段来搜索。如果你傻乎乎往 ES 里写入一行数据所有的字段就会导致说 90% 的数据是不用来搜索的结果硬是占据了 ES 机器上的 filesystem cache 的空间单条数据的数据量越大就会导致 filesystem cahce 能缓存的数据就越少。其实仅仅写入 ES 中要用来检索的少数几个字段就可以了比如说就写入 ES id,name,age 三个字段然后你可以把其他的字段数据存在 mysql/hbase 里我们一般是建议用 ES hbase 这么一个架构。
写入 ES 的数据最好小于等于或者是略微大于 ES 的 filesystem cache 的内存容量。然后你从 ES 检索可能就花费 20ms然后再根据 ES 返回的 id 去 hbase 里查询查 20 条数据可能也就耗费个 30ms可能你原来那么玩儿1T 数据都放 es会每次查询都是 5~10s现在可能性能就会很高每次查询就是 50ms。
数据预热
假如说哪怕是你就按照上述的方案去做了ES 集群中每个机器写入的数据量还是超过了 filesystem cache 一倍比如说你写入一台机器 60G 数据结果 filesystem cache 就 30G还是有 30G 数据留在了磁盘上。
其实可以做数据预热。
举个例子拿微博来说你可以把一些大 V平时看的人很多的数据你自己提前后台搞个系统每隔一会儿自己的后台系统去搜索一下热数据刷到 filesystem cache 里去后面用户实际上来看这个热数据的时候他们就是直接从内存里搜索了很快。
或者是电商你可以将平时查看最多的一些商品比如说 iphone 8热数据提前后台搞个程序每隔 1 分钟自己主动访问一次刷到 filesystem cache 里去。
对于那些你觉得比较热的、经常会有人访问的数据最好做一个专门的缓存预热子系统就是对热数据每隔一段时间就提前访问一下让数据进入 filesystem cache 里面去。这样下次别人访问的时候性能一定会好很多。
冷热分离
ES 可以做类似于 mysql 的水平拆分就是说将大量的访问很少、频率很低的数据单独写一个索引然后将访问很频繁的热数据单独写一个索引。最好是将冷数据写入一个索引中然后热数据写入另外一个索引中这样可以确保热数据在被预热之后尽量都让他们留在 filesystem os cache 里别让冷数据给冲刷掉。
假设你有 6 台机器2 个索引一个放冷数据一个放热数据每个索引 3 个 shard。3 台机器放热数据 index另外 3 台机器放冷数据 index。然后这样的话你大量的时间是在访问热数据 index热数据可能就占总数据量的 10%此时数据量很少几乎全都保留在 filesystem cache 里面了就可以确保热数据的访问性能是很高的。但是对于冷数据而言是在别的 index 里的跟热数据 index 不在相同的机器上大家互相之间都没什么联系了。如果有人访问冷数据可能大量数据是在磁盘上的此时性能差点就 10% 的人去访问冷数据90% 的人在访问热数据也无所谓了。
document模型设计
对于 MySQL我们经常有一些复杂的关联查询。在 ES 里该怎么玩儿ES 里面的复杂的关联查询尽量别用一旦用了性能一般都不太好。
最好是先在 Java 系统里就完成关联将关联好的数据直接写入 ES 中。搜索的时候就不需要利用 ES 的搜索语法来完成 join 之类的关联搜索了。
document 模型设计是非常重要的很多操作不要在搜索的时候才想去执行各种复杂的乱七八糟的操作。ES 能支持的操作就那么多不要考虑用 ES 做一些它不好操作的事情。如果真的有那种操作尽量在 document 模型设计的时候写入的时候就完成。另外对于一些太复杂的操作比如 join/nested/parent-child 搜索都要尽量避免性能都很差的。
