elasticsearch index 之 put mapping

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
简介:

elasticsearch index 之 put mapping

 

mapping机制使得elasticsearch索引数据变的更加灵活,近乎于no schema。mapping可以在建立索引时设置,也可以在后期设置。后期设置可以是修改mapping(无法对已有的field属性进行修改,一般来说只是增加新的field)或者对没有mapping的索引设置mapping。put mapping操作必须是master节点来完成,因为它涉及到集群matedata的修改,同时它跟index和type密切相关。修改只是针对特定index的特定type。

Action support分析中我们分析过几种Action的抽象类型,put mapping Action属于TransportMasterNodeOperationAction的子类。它实现了masterOperation方法,每个继承自TransportMasterNodeOperationAction的子类都会根据自己的具体功能来实现这个方法。这里的实现如下所示:

复制代码
复制代码
    protected void masterOperation(final PutMappingRequest request, final ClusterState state, final ActionListener<PutMappingResponse> listener) throws ElasticsearchException {
        final String[] concreteIndices = clusterService.state().metaData().concreteIndices(request.indicesOptions(), request.indices());
//构造request PutMappingClusterStateUpdateRequest updateRequest = new PutMappingClusterStateUpdateRequest() .ackTimeout(request.timeout()).masterNodeTimeout(request.masterNodeTimeout()) .indices(concreteIndices).type(request.type()) .source(request.source()).ignoreConflicts(request.ignoreConflicts());       //调用putMapping方法,同时传入一个Listener metaDataMappingService.putMapping(updateRequest, new ActionListener<ClusterStateUpdateResponse>() { @Override public void onResponse(ClusterStateUpdateResponse response) { listener.onResponse(new PutMappingResponse(response.isAcknowledged())); } @Override public void onFailure(Throwable t) { logger.debug("failed to put mappings on indices [{}], type [{}]", t, concreteIndices, request.type()); listener.onFailure(t); } }); }
复制代码
复制代码

以上是TransportPutMappingAction对masterOperation方法的实现,这里并没有多少复杂的逻辑和操作。具体操作在matedataMappingService中。跟之前的CreateIndex一样,put Mapping也是向master提交一个updateTask。所有逻辑也都在execute方法中。这个task的基本跟CreateIndex一样,也需要在给定的时间内响应。它的代码如下所示:

复制代码
复制代码
 public void putMapping(final PutMappingClusterStateUpdateRequest request, final ActionListener<ClusterStateUpdateResponse> listener) {
    //提交一个高基本的updateTask
        clusterService.submitStateUpdateTask("put-mapping [" + request.type() + "]", Priority.HIGH, new AckedClusterStateUpdateTask<ClusterStateUpdateResponse>(request, listener) {

            @Override
            protected ClusterStateUpdateResponse newResponse(boolean acknowledged) {
                return new ClusterStateUpdateResponse(acknowledged);
            }

            @Override
            public ClusterState execute(final ClusterState currentState) throws Exception {
                List<String> indicesToClose = Lists.newArrayList();
                try {
//必须针对已经在matadata中存在的index,否则抛出异常 for (String index : request.indices()) { if (!currentState.metaData().hasIndex(index)) { throw new IndexMissingException(new Index(index)); } } //还需要存在于indices中,否则无法进行操作。所以这里要进行预建 for (String index : request.indices()) { if (indicesService.hasIndex(index)) { continue; } final IndexMetaData indexMetaData = currentState.metaData().index(index);
//不存在就进行创建 IndexService indexService = indicesService.createIndex(indexMetaData.index(), indexMetaData.settings(), clusterService.localNode().id()); indicesToClose.add(indexMetaData.index()); // make sure to add custom default mapping if exists if (indexMetaData.mappings().containsKey(MapperService.DEFAULT_MAPPING)) { indexService.mapperService().merge(MapperService.DEFAULT_MAPPING, indexMetaData.mappings().get(MapperService.DEFAULT_MAPPING).source(), false); } // only add the current relevant mapping (if exists) if (indexMetaData.mappings().containsKey(request.type())) { indexService.mapperService().merge(request.type(), indexMetaData.mappings().get(request.type()).source(), false); } }             //合并更新Mapping Map<String, DocumentMapper> newMappers = newHashMap(); Map<String, DocumentMapper> existingMappers = newHashMap();
//针对每个index进行Mapping合并 for (String index : request.indices()) { IndexService indexService = indicesService.indexServiceSafe(index); // try and parse it (no need to add it here) so we can bail early in case of parsing exception DocumentMapper newMapper; DocumentMapper existingMapper = indexService.mapperService().documentMapper(request.type()); if (MapperService.DEFAULT_MAPPING.equals(request.type())) {//存在defaultmapping则合并default mapping // _default_ types do not go through merging, but we do test the new settings. Also don't apply the old default newMapper = indexService.mapperService().parse(request.type(), new CompressedString(request.source()), false); } else { newMapper = indexService.mapperService().parse(request.type(), new CompressedString(request.source()), existingMapper == null); if (existingMapper != null) { // first, simulate DocumentMapper.MergeResult mergeResult = existingMapper.merge(newMapper, mergeFlags().simulate(true)); // if we have conflicts, and we are not supposed to ignore them, throw an exception if (!request.ignoreConflicts() && mergeResult.hasConflicts()) { throw new MergeMappingException(mergeResult.conflicts()); } } } newMappers.put(index, newMapper); if (existingMapper != null) { existingMappers.put(index, existingMapper); } } String mappingType = request.type(); if (mappingType == null) { mappingType = newMappers.values().iterator().next().type(); } else if (!mappingType.equals(newMappers.values().iterator().next().type())) { throw new InvalidTypeNameException("Type name provided does not match type name within mapping definition"); } if (!MapperService.DEFAULT_MAPPING.equals(mappingType) && !PercolatorService.TYPE_NAME.equals(mappingType) && mappingType.charAt(0) == '_') { throw new InvalidTypeNameException("Document mapping type name can't start with '_'"); } final Map<String, MappingMetaData> mappings = newHashMap(); for (Map.Entry<String, DocumentMapper> entry : newMappers.entrySet()) { String index = entry.getKey(); // do the actual merge here on the master, and update the mapping source DocumentMapper newMapper = entry.getValue(); IndexService indexService = indicesService.indexService(index); if (indexService == null) { continue; } CompressedString existingSource = null; if (existingMappers.containsKey(entry.getKey())) { existingSource = existingMappers.get(entry.getKey()).mappingSource(); } DocumentMapper mergedMapper = indexService.mapperService().merge(newMapper.type(), newMapper.mappingSource(), false); CompressedString updatedSource = mergedMapper.mappingSource(); if (existingSource != null) { if (existingSource.equals(updatedSource)) { // same source, no changes, ignore it } else { // use the merged mapping source mappings.put(index, new MappingMetaData(mergedMapper)); if (logger.isDebugEnabled()) { logger.debug("[{}] update_mapping [{}] with source [{}]", index, mergedMapper.type(), updatedSource); } else if (logger.isInfoEnabled()) { logger.info("[{}] update_mapping [{}]", index, mergedMapper.type()); } } } else { mappings.put(index, new MappingMetaData(mergedMapper)); if (logger.isDebugEnabled()) { logger.debug("[{}] create_mapping [{}] with source [{}]", index, newMapper.type(), updatedSource); } else if (logger.isInfoEnabled()) { logger.info("[{}] create_mapping [{}]", index, newMapper.type()); } } } if (mappings.isEmpty()) { // no changes, return return currentState; }             //根据mapping的更新情况重新生成matadata MetaData.Builder builder = MetaData.builder(currentState.metaData()); for (String indexName : request.indices()) { IndexMetaData indexMetaData = currentState.metaData().index(indexName); if (indexMetaData == null) { throw new IndexMissingException(new Index(indexName)); } MappingMetaData mappingMd = mappings.get(indexName); if (mappingMd != null) { builder.put(IndexMetaData.builder(indexMetaData).putMapping(mappingMd)); } } return ClusterState.builder(currentState).metaData(builder).build(); } finally { for (String index : indicesToClose) { indicesService.removeIndex(index, "created for mapping processing"); } } } }); }
复制代码
复制代码

 以上就是mapping的设置过程,首先它跟Create index一样,只有master节点才能操作,而且是以task的形式提交给master;其次它的本质是将request中的mapping和index现存的或者是default mapping合并,并最终生成新的matadata更新到集群的各个节点。

总结:集群中的master操作无论是index方面还是集群方面,最终都是集群matadata的更新过程。而这些操作只能在master上进行,并且都是会超时的任务。put mapping当然也不例外。上面的两段代码基本概况了mapping的设置过程。这里就不再重复了。这里还有一个问题没有涉及到就是mapping的合并。mapping合并会在很多地方用到。

 

转自:http://www.cnblogs.com/zziawanblog/p/7011367.html












本文转自张昺华-sky博客园博客,原文链接:http://www.cnblogs.com/bonelee/p/7382326.html,如需转载请自行联系原作者

相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
相关文章
|
3月前
|
存储 API 数据库
检索服务elasticsearch索引(Index)
【8月更文挑战第23天】
66 6
|
5月前
|
存储
Elasticsearch exception [type=cluster_block_exception, reason=blocked by: [FORBIDDEN/12/index r【已解决】
Elasticsearch exception [type=cluster_block_exception, reason=blocked by: [FORBIDDEN/12/index r【已解决】
138 1
|
5月前
|
存储 自然语言处理 关系型数据库
Elasticsearch如何修改Mapping结构并实现业务零停机
Elasticsearch如何修改Mapping结构并实现业务零停机
|
5月前
|
存储 JSON 定位技术
深入理解Elasticsearch的索引映射(mapping)
深入理解Elasticsearch的索引映射(mapping)
|
6月前
|
存储 API 索引
Elasticsearch 8.X 防止 Mapping “爆炸”的三种方案
Elasticsearch 8.X 防止 Mapping “爆炸”的三种方案
76 0
|
6月前
|
索引
Elasticsearch exception [type=illegal_argument_exception, reason=index [.1] is the write index for data stream [slowlog] and cannot be deleted]
在 Elasticsearch 中,你尝试删除的索引是一个数据流(data stream)的一部分,而且是数据流的写入索引(write index),因此无法直接删除它。为了解决这个问题,你可以按照以下步骤进行操作:
357 0
|
6月前
|
API 索引
Elasticsearch Index Shard Allocation 索引分片分配策略
Elasticsearch Index Shard Allocation 索引分片分配策略
139 1
|
6月前
|
JSON 自然语言处理 定位技术
Elasticsearch Mapping是啥?
Elasticsearch Mapping是啥?
108 0
|
存储 JSON 自然语言处理
【Elasticsearch】学好Elasticsearch系列-Mapping
【Elasticsearch】学好Elasticsearch系列-Mapping
135 0
|
API
Elasticsearch - cluster_block_exception [FORBIDDEN/12/index read-only / allow delete (api)]
Elasticsearch - cluster_block_exception [FORBIDDEN/12/index read-only / allow delete (api)]
124 0
Elasticsearch - cluster_block_exception [FORBIDDEN/12/index read-only / allow delete (api)]