elasticsearch索引創(chuàng)建create?index集群matedata更新
更新時間：2022年04月22日 11:27:50 作者：zziawan
這篇文章主要介紹了elasticsearch索引創(chuàng)建create?index及集群matedata更新，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪
創(chuàng)建索引更新集群index matedata
創(chuàng)建索引需要創(chuàng)建索引并且更新集群index matedata，這一過程在MetaDataCreateIndexService的createIndex方法中完成。這里會提交一個高優(yōu)先級，AckedClusterStateUpdateTask類型的task。索引創(chuàng)建需要即時得到反饋，異常這個task需要返回，會超時，而且這個任務(wù)的優(yōu)先級也非常高。
下面具體看一下它的execute方法，這個方法會在master執(zhí)行任務(wù)時調(diào)用，這個方法非常長，主要完成以下三個功能：更新合并request，template中的mapping和setting，調(diào)用indiceService創(chuàng)建索引，對創(chuàng)建后的索引添加mapping。這一系列功能完成后，合并完成后生成新的matedata，并更新集群狀態(tài)，完成了索引的創(chuàng)建。具體的調(diào)用方法參考上一篇。
首先創(chuàng)建index的create方法
代碼如下所示：
@Override
public ClusterState execute(ClusterState currentState) throws Exception {
boolean indexCreated = false;
String removalReason = null;
try {
//檢查request的合法性，1.5版本主要檢查index名字是否合法，如不能含有某些字符，另外就是集群節(jié)點(diǎn)版本
validate(request, currentState);
for (Alias alias : request.aliases()) {//檢查別稱是否合法
aliasValidator.validateAlias(alias, request.index(), currentState.metaData());
}
// 查找索引模板
List<IndexTemplateMetaData> templates = findTemplates(request, currentState, indexTemplateFilter);
Map<String, Custom> customs = Maps.newHashMap();
// add the request mapping
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)建索引的時候還是會附帶著mapping，在request中mapping是一個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 (IndexTemplateMetaData template : 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()) {
String type = cursor.key;
IndexMetaData.Custom custom = cursor.value;
IndexMetaData.Custom existing = customs.get(type);
if (existing == null) {
customs.put(type, custom);
} else {
IndexMetaData.Custom merged = IndexMetaData.lookupFactorySafe(type).merge(existing, custom);
customs.put(type, merged);
}
}
//處理合并別名
for (ObjectObjectCursor<String, AliasMetaData> cursor : template.aliases()) {
AliasMetaData aliasMetaData = cursor.value;
//if an alias with same name came with the create index request itself,
// ignore this one taken from the index template
if (request.aliases().contains(new Alias(aliasMetaData.alias()))) {
continue;
}
//if an alias with same name was already processed, ignore this one
if (templatesAliases.containsKey(cursor.key)) {
continue;
}
//Allow templatesAliases to be templated by replacing a token with the name of the index that we are applying it to
if (aliasMetaData.alias().contains("{index}")) {
String templatedAlias = 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)在開始處理配置基本的mapping，合并邏輯跟之前相同，只是mapping來源不同
File mappingsDir = new File(environment.configFile(), "mappings");
if (mappingsDir.isDirectory()) {
// first index level
File indexMappingsDir = new File(mappingsDir, request.index());
if (indexMappingsDir.isDirectory()) {
addMappings(mappings, indexMappingsDir);
}
// second is the _default mapping
File defaultMappingsDir = new File(mappingsDir, "_default");
if (defaultMappingsDir.isDirectory()) {
addMappings(mappings, defaultMappingsDir);
}
}
//處理index的配置（setting）
ImmutableSettings.Builder indexSettingsBuilder = settingsBuilder();
//加入模板中的setting
for (int i = templates.size() - 1; i >= 0; i--) {
indexSettingsBuilder.put(templates.get(i).settings());
}
// 加入request中的mapping，request中設(shè)置會覆蓋模板中的設(shè)置
indexSettingsBuilder.put(request.settings());
//處理shard，shard數(shù)量不能小于1，因此這里需要特殊處理，如果沒有則要使用默認(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) {
DiscoveryNodes nodes = currentState.nodes();
final Version createdVersion = 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
Settings actualIndexSettings = indexSettingsBuilder.build();
// 通過indiceservice創(chuàng)建索引
indicesService.createIndex(request.index(), actualIndexSettings, clusterService.localNode().id());
indexCreated = true;
//如果創(chuàng)建成功這里就可以獲取到對應(yīng)的indexservice，否則會拋出異常
IndexService indexService = indicesService.indexServiceSafe(request.index());
//獲取mappingService試圖放置mapping
MapperService mapperService = indexService.mapperService();
// 為索引添加mapping，首先是默認(rèn)mapping
if (mappings.containsKey(MapperService.DEFAULT_MAPPING)) {
try {
mapperService.merge(MapperService.DEFAULT_MAPPING, new CompressedString(XContentFactory.jsonBuilder().map(mappings.get(MapperService.DEFAULT_MAPPING)).string()), false);
} catch (Exception e) {
removalReason = "failed on parsing default mapping on index creation";
throw new MapperParsingException("mapping [" + MapperService.DEFAULT_MAPPING + "]", e);
}
}
for (Map.Entry<String, Map<String, Object>> entry : mappings.entrySet()) {
if (entry.getKey().equals(MapperService.DEFAULT_MAPPING)) {
continue;
}
try {
// apply the default here, its the first time we parse it
mapperService.merge(entry.getKey(), new CompressedString(XContentFactory.jsonBuilder().map(entry.getValue()).string()), true);
} catch (Exception e) {
removalReason = "failed on parsing mappings on index creation";
throw new MapperParsingException("mapping [" + entry.getKey() + "]", e);
}
}
//添加request中的別稱
IndexQueryParserService indexQueryParserService = indexService.queryParserService();
for (Alias alias : request.aliases()) {
if (Strings.hasLength(alias.filter())) {
aliasValidator.validateAliasFilter(alias.name(), alias.filter(), indexQueryParserService);
}
}
for (AliasMetaData aliasMetaData : templatesAliases.values()) {
if (aliasMetaData.filter() != null) {
aliasValidator.validateAliasFilter(aliasMetaData.alias(), aliasMetaData.filter().uncompressed(), indexQueryParserService);
}
}
// 以下更新Index的matedata，
Map<String, MappingMetaData> mappingsMetaData = Maps.newHashMap();
for (DocumentMapper mapper : mapperService.docMappers(true)) {
MappingMetaData mappingMd = new MappingMetaData(mapper);
mappingsMetaData.put(mapper.type(), mappingMd);
}
final IndexMetaData.Builder indexMetaDataBuilder = IndexMetaData.builder(request.index()).settings(actualIndexSettings);
for (MappingMetaData mappingMd : mappingsMetaData.values()) {
indexMetaDataBuilder.putMapping(mappingMd);
}
for (AliasMetaData aliasMetaData : templatesAliases.values()) {
indexMetaDataBuilder.putAlias(aliasMetaData);
}
for (Alias alias : request.aliases()) {
AliasMetaData aliasMetaData = 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
final IndexMetaData indexMetaData;
try {
indexMetaData = indexMetaDataBuilder.build();
} catch (Exception e) {
removalReason = "failed to build index metadata";
throw e;
}
indexService.indicesLifecycle().beforeIndexAddedToCluster(new Index(request.index()),
indexMetaData.settings());
//更新集群的matedata，將新build的indexmatadata加入到metadata中
MetaData newMetaData = MetaData.builder(currentState.metaData())
.put(indexMetaData, false)
.build();
logger.info("[{}] creating index, cause [{}], templates {}, shards [{}]/[{}], mappings {}", request.index(), request.cause(), templateNames, indexMetaData.numberOfShards(), indexMetaData.numberOfReplicas(), mappings.keySet());
//阻塞集群，更新matadata
ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
if (!request.blocks().isEmpty()) {
for (ClusterBlock block : request.blocks()) {
blocks.addIndexBlock(request.index(), block);
}
}
if (request.state() == State.CLOSE) {
blocks.addIndexBlock(request.index(), MetaDataIndexStateService.INDEX_CLOSED_BLOCK);
}
ClusterState updatedState = ClusterState.builder(currentState).blocks(blocks).metaData(newMetaData).build();
if (request.state() == State.OPEN) {
RoutingTable.Builder routingTableBuilder = RoutingTable.builder(updatedState.routingTable())
.addAsNew(updatedState.metaData().index(request.index()));
RoutingAllocation.Result routingResult = allocationService.reroute(ClusterState.builder(updatedState).routingTable(routingTableBuilder).build());
updatedState = ClusterState.builder(updatedState).routingResult(routingResult).build();
}
removalReason = "cleaning up after validating index on master";
return updatedState;
} finally {
if (indexCreated) {
// Index was already partially created - need to clean up
indicesService.removeIndex(request.index(), removalReason != null ? removalReason : "failed to create index");
}
}
}
});
}以上就是創(chuàng)建index的create方法，方法中主要進(jìn)行了兩個動作：合并更新index的matadata和創(chuàng)建index。更新合并matadata的過程都在上面的代碼中體現(xiàn)了。
從indice中獲取對應(yīng)的IndexService
創(chuàng)建索引是調(diào)用indiceSerivice構(gòu)建一個guice的injector，這個injector包含了Index的所有功能（如分詞，相似度等）。同時會將其存儲到indiceService中，以一個map的格式存儲Map<String, Tuple<IndexService, Injector>> indices。運(yùn)行中的集群每次對某個索引的操作都首先從indice中獲取對應(yīng)的IndexService。
這一部分代碼如下所示：
public synchronized IndexService createIndex(String sIndexName, @IndexSettings Settings settings, String localNodeId) throws ElasticsearchException {
if (!lifecycle.started()) {
throw new ElasticsearchIllegalStateException("Can't create an index [" + sIndexName + "], node is closed");
}
Index index = new Index(sIndexName);
//檢測index是否已經(jīng)存在
if (indices.containsKey(index.name())) {
throw new IndexAlreadyExistsException(index);
}
indicesLifecycle.beforeIndexCreated(index, settings);
logger.debug("creating Index [{}], shards [{}]/[{}]", sIndexName, settings.get(SETTING_NUMBER_OF_SHARDS), settings.get(SETTING_NUMBER_OF_REPLICAS));
Settings indexSettings = settingsBuilder()
.put(this.settings)
.put(settings)
.classLoader(settings.getClassLoader())
.build();
//構(gòu)建index對應(yīng)的injector
ModulesBuilder modules = new ModulesBuilder();
modules.add(new IndexNameModule(index));
modules.add(new LocalNodeIdModule(localNodeId));
modules.add(new IndexSettingsModule(index, indexSettings));
modules.add(new IndexPluginsModule(indexSettings, pluginsService));
modules.add(new IndexStoreModule(indexSettings));
modules.add(new IndexEngineModule(indexSettings));
modules.add(new AnalysisModule(indexSettings, indicesAnalysisService));
modules.add(new SimilarityModule(indexSettings));
modules.add(new IndexCacheModule(indexSettings));
modules.add(new IndexFieldDataModule(indexSettings));
modules.add(new CodecModule(indexSettings));
modules.add(new MapperServiceModule());
modules.add(new IndexQueryParserModule(indexSettings));
modules.add(new IndexAliasesServiceModule());
modules.add(new IndexGatewayModule(indexSettings, injector.getInstance(Gateway.class)));
modules.add(new IndexModule(indexSettings));
Injector indexInjector;
try {
indexInjector = modules.createChildInjector(injector);
} catch (CreationException e) {
throw new IndexCreationException(index, Injectors.getFirstErrorFailure(e));
} catch (Throwable e) {
throw new IndexCreationException(index, e);
}
IndexService indexService = indexInjector.getInstance(IndexService.class);
indicesLifecycle.afterIndexCreated(indexService);
//將Indexservice和IndexInjector加入到indice map中
indices = newMapBuilder(indices).put(index.name(), new Tuple<>(indexService, indexInjector)).immutableMap();
return indexService;
}以上方法就是具體創(chuàng)建索引的過程，它是在master上操作的，同時它是同步方法。這樣才能保證集群的Index創(chuàng)建一致性，因此這也會導(dǎo)致之前所說的大量創(chuàng)建創(chuàng)建索引時候的速度瓶頸。但是創(chuàng)建大量索引的動作是不常見的，需要盡量避免。創(chuàng)建一個索引對于一個集群來說就是開啟對于該索引的各種操作，因此這里通過guice將索引的各個功能模塊注入，并獲得index操作的接口類Indexservice。如果這個方法執(zhí)行成功，則可以合并template及request中的mapping，并且向剛創(chuàng)建的索引添加合并后的mapping，最后構(gòu)建新的matadata，并將集群新的matadata發(fā)送給各個節(jié)點(diǎn)完成索引創(chuàng)建。
總結(jié)
索引創(chuàng)建的過程包括三步：更新集群matadata，調(diào)用indiceService中創(chuàng)建索引，向新創(chuàng)建的索引中放置（合并到Index對應(yīng)的mappingService中）mapping。這三步都在以上的兩個方法中。完成這三步，集群中就保存了新索引的信息，同時索引配置和mapping放置也完成。索引就可以正常使用。
以上就是elasticsearch索引創(chuàng)建create index集群matedata更新的詳細(xì)內(nèi)容，更多關(guān)于elasticsearch索引create index matedata更新的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
這篇文章主要為大家介紹了mybatis的Interceptor機(jī)制源碼解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-08-08
今天帶大家了解SpringBoot環(huán)境配置的相關(guān)知識,文中有非常詳細(xì)的代碼示例,對正在學(xué)習(xí)java的小伙伴們有很好地幫助,需要的朋友可以參考下2021-05-05
Log4j 配置日志打印時區(qū)的實(shí)現(xiàn)方法
下面小編就為大家分享一篇Log4j 配置日志打印時區(qū)的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2017-12-12
Java8中l(wèi)ambda表達(dá)式的應(yīng)用及一些泛型相關(guān)知識
這篇文章主要介紹了Java8中l(wèi)ambda表達(dá)式的應(yīng)用及一些泛型相關(guān)知識的相關(guān)資料2017-01-01
這篇文章主要介紹了ArrayList底層操作機(jī)制源碼解析,當(dāng)創(chuàng)建ArrayList對象時,如果使用的是無參構(gòu)造器,則初始elementData容量為0,第1次添加,則擴(kuò)容elementData為10,如需要再次擴(kuò)容,則擴(kuò)容elementData為1.5倍,需要的朋友可以參考下2023-09-09
我知道很多朋友都和我一樣,在JAVA程序中似乎經(jīng)常見到this,自己也偶爾用到它,但是到底this該怎么用,卻心中無數(shù),下面這篇文章主要給大家介紹了關(guān)于Java中this關(guān)鍵字用法的相關(guān)資料,需要的朋友可以參考下2023-05-05
詳解java.lang.NumberFormatException錯誤及解決辦法
這篇文章主要介紹了詳解java.lang.NumberFormatException錯誤及解決辦法，本文詳解的介紹了錯誤的解決方法，感興趣的可以一起來了解一下2020-05-05
Java復(fù)制(拷貝)數(shù)組的4種方法：arraycopy()方法、clone() 方法、copyOf()和copyOfRa
這篇文章主要介紹了Java復(fù)制(拷貝)數(shù)組的4種方法：arraycopy()方法、clone() 方法、copyOf()和copyOfRan，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-01-01
Springboot如何切換默認(rèn)的Tomcat容器
這篇文章主要介紹了Springboot如何切換默認(rèn)的Tomcat容器,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-06-06
使用shardingsphere對SQLServer坑的解決
本文主要介紹了使用shardingsphere對SQLServer坑的解決，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-03-03