mapping機制使得elasticsearch索引數據變的更加靈活,近乎於no schema。mapping能夠在創建索引時設置,也能夠在後期設置。後期設置能夠是修改mapping(沒法對已有的field屬性進行修改,通常來講只是增長新的field)或者對沒有mapping的索引設置mapping。put mapping操做必須是master節點來完成,由於它涉及到集羣matedata的修改,同時它跟index和type密切相關。修改只是針對特定index的特定type。html
在Action support分析中咱們分析過幾種Action的抽象類型,put mapping Action屬於TransportMasterNodeOperationAction的子類。它實現了masterOperation方法,每一個繼承自TransportMasterNodeOperationAction的子類都會根據本身的具體功能來實現這個方法。這裏的實現以下所示:app
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同樣,也須要在給定的時間內響應。它的代碼以下所示:elasticsearch
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更新到集羣的各個節點。ide
總結:集羣中的master操做不管是index方面仍是集羣方面,最終都是集羣matadata的更新過程。而這些操做只能在master上進行,而且都是會超時的任務。put mapping固然也不例外。上面的兩段代碼基本概況了mapping的設置過程。這裏就再也不重複了。這裏還有一個問題沒有涉及到就是mapping的合併。mapping合併會在不少地方用到。post
轉自:http://www.cnblogs.com/zziawanblog/p/7011367.htmlui