elasticsearch索引创建create index集群matedata更新的方法
生活随笔
收集整理的這篇文章主要介紹了
elasticsearch索引创建create index集群matedata更新的方法
小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.
本文小編為大家詳細(xì)介紹“elasticsearch索引創(chuàng)建createindex集群matedata更新的方法”，內(nèi)容詳細(xì)，步驟清晰，細(xì)節(jié)處理妥當(dāng)，希望這篇“elasticsearch索引創(chuàng)建createindex集群matedata更新的方法”文章能幫助大家解決疑惑，下面跟著小編的思路慢慢深入，一起來(lái)學(xué)習(xí)新知識(shí)吧。
更新集群index matedata
創(chuàng)建索引需要?jiǎng)?chuàng)建索引并且更新集群index matedata，這一過(guò)程在MetaDataCreateIndexService的createIndex方法中完成。這里會(huì)提交一個(gè)高優(yōu)先級(jí)，AckedClusterStateUpdateTask類(lèi)型的task。索引創(chuàng)建需要即時(shí)得到反饋，異常這個(gè)task需要返回，會(huì)超時(shí)，而且這個(gè)任務(wù)的優(yōu)先級(jí)也非常高。
下面具體看一下它的execute方法，這個(gè)方法會(huì)在master執(zhí)行任務(wù)時(shí)調(diào)用，這個(gè)方法非常長(zhǎng)，主要完成以下三個(gè)功能：更新合并request，template中的mapping和setting，調(diào)用indiceService創(chuàng)建索引，對(duì)創(chuàng)建后的索引添加mapping。這一系列功能完成后，合并完成后生成新的matedata，并更新集群狀態(tài)，完成了索引的創(chuàng)建。
首先創(chuàng)建index的create方法
代碼如下所示：
@Override
publicClusterStateexecute(ClusterStatecurrentState)throwsException{
booleanindexCreated=false;
StringremovalReason=null;
try{
//檢查request的合法性，1.5版本主要檢查index名字是否合法，如不能含有某些字符，另外就是集群節(jié)點(diǎn)版本
validate(request,currentState);
for(Aliasalias:request.aliases()){//檢查別稱(chēng)是否合法
aliasValidator.validateAlias(alias,request.index(),currentState.metaData());
}
//查找索引模板
List<IndexTemplateMetaData>templates=findTemplates(request,currentState,indexTemplateFilter);
Map<String,Custom>customs=Maps.newHashMap();
//addtherequestmapping
Map<String,Map<String,Object>>mappings=Maps.newHashMap();
Map<String,AliasMetaData>templatesAliases=Maps.newHashMap();
List<String>templateNames=Lists.newArrayList();
//取出request中的mapping配置，雖然mapping可以后面添加，多數(shù)情況創(chuàng)建索引的時(shí)候還是會(huì)附帶著mapping，在request中mapping是一個(gè)map
for(Map.Entry<String,String>entry:request.mappings().entrySet()){
mappings.put(entry.getKey(),parseMapping(entry.getValue()));
}
//一些預(yù)設(shè)如warm等
for(Map.Entry<String,Custom>entry:request.customs().entrySet()){
customs.put(entry.getKey(),entry.getValue());
}
//將找到的template和request中的mapping合并
for(IndexTemplateMetaDatatemplate:templates){
templateNames.add(template.getName());
for(ObjectObjectCursor<String,CompressedString>cursor:template.mappings()){
if(mappings.containsKey(cursor.key)){
XContentHelper.mergeDefaults(mappings.get(cursor.key),parseMapping(cursor.value.string()));
}else{
mappings.put(cursor.key,parseMapping(cursor.value.string()));
}
}
//合并custom
for(ObjectObjectCursor<String,Custom>cursor:template.customs()){
Stringtype=cursor.key;
IndexMetaData.Customcustom=cursor.value;
IndexMetaData.Customexisting=customs.get(type);
if(existing==null){
customs.put(type,custom);
}else{
IndexMetaData.Custommerged=IndexMetaData.lookupFactorySafe(type).merge(existing,custom);
customs.put(type,merged);
}
}
//處理合并別名
for(ObjectObjectCursor<String,AliasMetaData>cursor:template.aliases()){
AliasMetaDataaliasMetaData=cursor.value;
//ifanaliaswithsamenamecamewiththecreateindexrequestitself,
//ignorethisonetakenfromtheindextemplate
if(request.aliases().contains(newAlias(aliasMetaData.alias()))){
continue;
}
//ifanaliaswithsamenamewasalreadyprocessed,ignorethisone
if(templatesAliases.containsKey(cursor.key)){
continue;
}
//AllowtemplatesAliasestobetemplatedbyreplacingatokenwiththenameoftheindexthatweareapplyingitto
if(aliasMetaData.alias().contains("{index}")){
StringtemplatedAlias=aliasMetaData.alias().replace("{index}",request.index());
aliasMetaData=AliasMetaData.newAliasMetaData(aliasMetaData,templatedAlias);
}
aliasValidator.validateAliasMetaData(aliasMetaData,request.index(),currentState.metaData());
templatesAliases.put(aliasMetaData.alias(),aliasMetaData);
}
}
//合并完template和request，現(xiàn)在開(kāi)始處理配置基本的mapping，合并邏輯跟之前相同，只是mapping來(lái)源不同
FilemappingsDir=newFile(environment.configFile(),"mappings");
if(mappingsDir.isDirectory()){
//firstindexlevel
FileindexMappingsDir=newFile(mappingsDir,request.index());
if(indexMappingsDir.isDirectory()){
addMappings(mappings,indexMappingsDir);
}
//secondisthe_defaultmapping
FiledefaultMappingsDir=newFile(mappingsDir,"_default");
if(defaultMappingsDir.isDirectory()){
addMappings(mappings,defaultMappingsDir);
}
}
//處理index的配置（setting）
ImmutableSettings.BuilderindexSettingsBuilder=settingsBuilder();
//加入模板中的setting
for(inti=templates.size()-1;i>=0;i--){
indexSettingsBuilder.put(templates.get(i).settings());
}
//加入request中的mapping，request中設(shè)置會(huì)覆蓋模板中的設(shè)置
indexSettingsBuilder.put(request.settings());
//處理shard，shard數(shù)量不能小于1，因此這里需要特殊處理，如果沒(méi)有則要使用默認(rèn)值
if(request.index().equals(ScriptService.SCRIPT_INDEX)){
indexSettingsBuilder.put(SETTING_NUMBER_OF_SHARDS,settings.getAsInt(SETTING_NUMBER_OF_SHARDS,1));
}else{
if(indexSettingsBuilder.get(SETTING_NUMBER_OF_SHARDS)==null){
if(request.index().equals(riverIndexName)){
indexSettingsBuilder.put(SETTING_NUMBER_OF_SHARDS,settings.getAsInt(SETTING_NUMBER_OF_SHARDS,1));
}else{
indexSettingsBuilder.put(SETTING_NUMBER_OF_SHARDS,settings.getAsInt(SETTING_NUMBER_OF_SHARDS,5));
}
}
}
if(request.index().equals(ScriptService.SCRIPT_INDEX)){
indexSettingsBuilder.put(SETTING_NUMBER_OF_REPLICAS,settings.getAsInt(SETTING_NUMBER_OF_REPLICAS,0));
indexSettingsBuilder.put(SETTING_AUTO_EXPAND_REPLICAS,"0-all");
}
else{
if(indexSettingsBuilder.get(SETTING_NUMBER_OF_REPLICAS)==null){
if(request.index().equals(riverIndexName)){
indexSettingsBuilder.put(SETTING_NUMBER_OF_REPLICAS,settings.getAsInt(SETTING_NUMBER_OF_REPLICAS,1));
}else{
indexSettingsBuilder.put(SETTING_NUMBER_OF_REPLICAS,settings.getAsInt(SETTING_NUMBER_OF_REPLICAS,1));
}
}
}
//處理副本
if(settings.get(SETTING_AUTO_EXPAND_REPLICAS)!=null&&indexSettingsBuilder.get(SETTING_AUTO_EXPAND_REPLICAS)==null){
indexSettingsBuilder.put(SETTING_AUTO_EXPAND_REPLICAS,settings.get(SETTING_AUTO_EXPAND_REPLICAS));
}
if(indexSettingsBuilder.get(SETTING_VERSION_CREATED)==null){
DiscoveryNodesnodes=currentState.nodes();
finalVersioncreatedVersion=Version.smallest(version,nodes.smallestNonClientNodeVersion());
indexSettingsBuilder.put(SETTING_VERSION_CREATED,createdVersion);
}
if(indexSettingsBuilder.get(SETTING_CREATION_DATE)==null){
indexSettingsBuilder.put(SETTING_CREATION_DATE,System.currentTimeMillis());
}
indexSettingsBuilder.put(SETTING_UUID,Strings.randomBase64UUID());
//創(chuàng)建setting
SettingsactualIndexSettings=indexSettingsBuilder.build();
//通過(guò)indiceservice創(chuàng)建索引
indicesService.createIndex(request.index(),actualIndexSettings,clusterService.localNode().id());
indexCreated=true;
//如果創(chuàng)建成功這里就可以獲取到對(duì)應(yīng)的indexservice，否則會(huì)拋出異常
IndexServiceindexService=indicesService.indexServiceSafe(request.index());
//獲取mappingService試圖放置mapping
MapperServicemapperService=indexService.mapperService();
//為索引添加mapping，首先是默認(rèn)mapping
if(mappings.containsKey(MapperService.DEFAULT_MAPPING)){
try{
mapperService.merge(MapperService.DEFAULT_MAPPING,newCompressedString(XContentFactory.jsonBuilder().map(mappings.get(MapperService.DEFAULT_MAPPING)).string()),false);
}catch(Exceptione){
removalReason="failedonparsingdefaultmappingonindexcreation";
thrownewMapperParsingException("mapping["+MapperService.DEFAULT_MAPPING+"]",e);
}
}
for(Map.Entry<String,Map<String,Object>>entry:mappings.entrySet()){
if(entry.getKey().equals(MapperService.DEFAULT_MAPPING)){
continue;
}
try{
//applythedefaulthere,itsthefirsttimeweparseit
mapperService.merge(entry.getKey(),newCompressedString(XContentFactory.jsonBuilder().map(entry.getValue()).string()),true);
}catch(Exceptione){
removalReason="failedonparsingmappingsonindexcreation";
thrownewMapperParsingException("mapping["+entry.getKey()+"]",e);
}
}
//添加request中的別稱(chēng)
IndexQueryParserServiceindexQueryParserService=indexService.queryParserService();
for(Aliasalias:request.aliases()){
if(Strings.hasLength(alias.filter())){
aliasValidator.validateAliasFilter(alias.name(),alias.filter(),indexQueryParserService);
}
}
for(AliasMetaDataaliasMetaData:templatesAliases.values()){
if(aliasMetaData.filter()!=null){
aliasValidator.validateAliasFilter(aliasMetaData.alias(),aliasMetaData.filter().uncompressed(),indexQueryParserService);
}
}
//以下更新Index的matedata，
Map<String,MappingMetaData>mappingsMetaData=Maps.newHashMap();
for(DocumentMappermapper:mapperService.docMappers(true)){
MappingMetaDatamappingMd=newMappingMetaData(mapper);
mappingsMetaData.put(mapper.type(),mappingMd);
}
finalIndexMetaData.BuilderindexMetaDataBuilder=IndexMetaData.builder(request.index()).settings(actualIndexSettings);
for(MappingMetaDatamappingMd:mappingsMetaData.values()){
indexMetaDataBuilder.putMapping(mappingMd);
}
for(AliasMetaDataaliasMetaData:templatesAliases.values()){
indexMetaDataBuilder.putAlias(aliasMetaData);
}
for(Aliasalias:request.aliases()){
AliasMetaDataaliasMetaData=AliasMetaData.builder(alias.name()).filter(alias.filter())
.indexRouting(alias.indexRouting()).searchRouting(alias.searchRouting()).build();
indexMetaDataBuilder.putAlias(aliasMetaData);
}
for(Map.Entry<String,Custom>customEntry:customs.entrySet()){
indexMetaDataBuilder.putCustom(customEntry.getKey(),customEntry.getValue());
}
indexMetaDataBuilder.state(request.state());
//matedata更新完畢，build新的matedata
finalIndexMetaDataindexMetaData;
try{
indexMetaData=indexMetaDataBuilder.build();
}catch(Exceptione){
removalReason="failedtobuildindexmetadata";
throwe;
}
indexService.indicesLifecycle().beforeIndexAddedToCluster(newIndex(request.index()),
indexMetaData.settings());
//更新集群的matedata，將新build的indexmatadata加入到metadata中
MetaDatanewMetaData=MetaData.builder(currentState.metaData())
.put(indexMetaData,false)
.build();
logger.info("[{}]creatingindex,cause[{}],templates{},shards[{}]/[{}],mappings{}",request.index(),request.cause(),templateNames,indexMetaData.numberOfShards(),indexMetaData.numberOfReplicas(),mappings.keySet());
//阻塞集群，更新matadata
ClusterBlocks.Builderblocks=ClusterBlocks.builder().blocks(currentState.blocks());
if(!request.blocks().isEmpty()){
for(ClusterBlockblock:request.blocks()){
blocks.addIndexBlock(request.index(),block);
}
}
if(request.state()==State.CLOSE){
blocks.addIndexBlock(request.index(),MetaDataIndexStateService.INDEX_CLOSED_BLOCK);
}
ClusterStateupdatedState=ClusterState.builder(currentState).blocks(blocks).metaData(newMetaData).build();
if(request.state()==State.OPEN){
RoutingTable.BuilderroutingTableBuilder=RoutingTable.builder(updatedState.routingTable())
.addAsNew(updatedState.metaData().index(request.index()));
RoutingAllocation.ResultroutingResult=allocationService.reroute(ClusterState.builder(updatedState).routingTable(routingTableBuilder).build());
updatedState=ClusterState.builder(updatedState).routingResult(routingResult).build();
}
removalReason="cleaningupaftervalidatingindexonmaster";
returnupdatedState;
}finally{
if(indexCreated){
//Indexwasalreadypartiallycreated-needtocleanup
indicesService.removeIndex(request.index(),removalReason!=null?removalReason:"failedtocreateindex");
}
}
}
});
}
以上就是創(chuàng)建index的create方法，方法中主要進(jìn)行了兩個(gè)動(dòng)作：合并更新index的matadata和創(chuàng)建index。更新合并matadata的過(guò)程都在上面的代碼中體現(xiàn)了。
從indice中獲取對(duì)應(yīng)的IndexService
創(chuàng)建索引是調(diào)用indiceSerivice構(gòu)建一個(gè)guice的injector，這個(gè)injector包含了Index的所有功能（如分詞，相似度等）。同時(shí)會(huì)將其存儲(chǔ)到indiceService中，以一個(gè)map的格式存儲(chǔ)Map<String, Tuple<IndexService, Injector>> indices。運(yùn)行中的集群每次對(duì)某個(gè)索引的操作都首先從indice中獲取對(duì)應(yīng)的IndexService。
這一部分代碼如下所示：
publicsynchronizedIndexServicecreateIndex(StringsIndexName,@IndexSettingsSettingssettings,StringlocalNodeId)throwsElasticsearchException{
if(!lifecycle.started()){
thrownewElasticsearchIllegalStateException("Can'tcreateanindex["+sIndexName+"],nodeisclosed");
}
Indexindex=newIndex(sIndexName);
//檢測(cè)index是否已經(jīng)存在
if(indices.containsKey(index.name())){
thrownewIndexAlreadyExistsException(index);
}
indicesLifecycle.beforeIndexCreated(index,settings);
logger.debug("creatingIndex[{}],shards[{}]/[{}]",sIndexName,settings.get(SETTING_NUMBER_OF_SHARDS),settings.get(SETTING_NUMBER_OF_REPLICAS));
SettingsindexSettings=settingsBuilder()
.put(this.settings)
.put(settings)
.classLoader(settings.getClassLoader())
.build();
//構(gòu)建index對(duì)應(yīng)的injector
ModulesBuildermodules=newModulesBuilder();
modules.add(newIndexNameModule(index));
modules.add(newLocalNodeIdModule(localNodeId));
modules.add(newIndexSettingsModule(index,indexSettings));
modules.add(newIndexPluginsModule(indexSettings,pluginsService));
modules.add(newIndexStoreModule(indexSettings));
modules.add(newIndexEngineModule(indexSettings));
modules.add(newAnalysisModule(indexSettings,indicesAnalysisService));
modules.add(newSimilarityModule(indexSettings));
modules.add(newIndexCacheModule(indexSettings));
modules.add(newIndexFieldDataModule(indexSettings));
modules.add(newCodecModule(indexSettings));
modules.add(newMapperServiceModule());
modules.add(newIndexQueryParserModule(indexSettings));
modules.add(newIndexAliasesServiceModule());
modules.add(newIndexGatewayModule(indexSettings,injector.getInstance(Gateway.class)));
modules.add(newIndexModule(indexSettings));
InjectorindexInjector;
try{
indexInjector=modules.createChildInjector(injector);
}catch(CreationExceptione){
thrownewIndexCreationException(index,Injectors.getFirstErrorFailure(e));
}catch(Throwablee){
thrownewIndexCreationException(index,e);
}
IndexServiceindexService=indexInjector.getInstance(IndexService.class);
indicesLifecycle.afterIndexCreated(indexService);
//將Indexservice和IndexInjector加入到indicemap中
indices=newMapBuilder(indices).put(index.name(),newTuple<>(indexService,indexInjector)).immutableMap();
returnindexService;
}
以上方法就是具體創(chuàng)建索引的過(guò)程，它是在master上操作的，同時(shí)它是同步方法。這樣才能保證集群的Index創(chuàng)建一致性，因此這也會(huì)導(dǎo)致之前所說(shuō)的大量創(chuàng)建創(chuàng)建索引時(shí)候的速度瓶頸。但是創(chuàng)建大量索引的動(dòng)作是不常見(jiàn)的，需要盡量避免。創(chuàng)建一個(gè)索引對(duì)于一個(gè)集群來(lái)說(shuō)就是開(kāi)啟對(duì)于該索引的各種操作，因此這里通過(guò)guice將索引的各個(gè)功能模塊注入，并獲得index操作的接口類(lèi)Indexservice。如果這個(gè)方法執(zhí)行成功，則可以合并template及request中的mapping，并且向剛創(chuàng)建的索引添加合并后的mapping，最后構(gòu)建新的matadata，并將集群新的matadata發(fā)送給各個(gè)節(jié)點(diǎn)完成索引創(chuàng)建。
總結(jié)
以上是生活随笔為你收集整理的elasticsearch索引创建create index集群matedata更新的方法的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇： oracle 052考试,Oracle
- 下一篇： win10内存优化方法有哪些