创建索引更新集群index matedata
创建索引需要创建索引并且更新集群index matedata，这一过程在MetaDataCreateIndexService的createIndex方法中完成。这里会提交一个高优先级，AckedClusterStateUpdateTask类型的task。索引创建需要即时得到反馈，异常这个task需要返回，会超时，而且这个任务的优先级也非常高。
下面具体看一下它的execute方法，这个方法会在master执行任务时调用，这个方法非常长，主要完成以下三个功能：更新合并request，template中的mapping和setting，调用indiceService创建索引，对创建后的索引添加mapping。这一系列功能完成后，合并完成后生成新的matedata，并更新集群状态，完成了索引的创建。具体的调用方法参考上一篇。
首先创建index的create方法
代码如下所示：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
|
@Override public ClusterState execute(ClusterState currentState) throws Exception { boolean indexCreated = false; String removalReason = null; try { //检查request的合法性，1.5版本主要检查index名字是否合法，如不能含有某些字符，另外就是集群节点版本 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可以后面添加，多数情况创建索引的时候还是会附带着mapping，在request中mapping是一个map for (Map.Entry<String, String> entry : request.mappings().entrySet()) { mappings.put(entry.getKey(), parseMapping(entry.getValue())); } //一些预设如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，现在开始处理配置基本的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中设置会覆盖模板中的设置 indexSettingsBuilder.put(request.settings()); //处理shard，shard数量不能小于1，因此这里需要特殊处理，如果没有则要使用默认值 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()); //创建setting Settings actualIndexSettings = indexSettingsBuilder.build(); // 通过indiceservice创建索引 indicesService.createIndex(request.index(), actualIndexSettings, clusterService.localNode().id()); indexCreated = true; //如果创建成功这里就可以获取到对应的indexservice，否则会抛出异常 IndexService indexService = indicesService.indexServiceSafe(request.index()); //获取mappingService试图放置mapping MapperService mapperService = indexService.mapperService(); // 为索引添加mapping，首先是默认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"); } } } }); } |
以上就是创建index的create方法，方法中主要进行了两个动作：合并更新index的matadata和创建index。更新合并matadata的过程都在上面的代码中体现了。
从indice中获取对应的IndexService
创建索引是调用indiceSerivice构建一个guice的injector，这个injector包含了Index的所有功能（如分词，相似度等）。同时会将其存储到indiceService中，以一个map的格式存储Map<String, Tuple<IndexService, Injector>> indices。运行中的集群每次对某个索引的操作都首先从indice中获取对应的IndexService。
这一部分代码如下所示：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
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是否已经存在 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(); //构建index对应的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; } |
以上方法就是具体创建索引的过程，它是在master上操作的，同时它是同步方法。这样才能保证集群的Index创建一致性，因此这也会导致之前所说的大量创建创建索引时候的速度瓶颈。但是创建大量索引的动作是不常见的，需要尽量避免。创建一个索引对于一个集群来说就是开启对于该索引的各种操作，因此这里通过guice将索引的各个功能模块注入，并获得index操作的接口类Indexservice。如果这个方法执行成功，则可以合并template及request中的mapping，并且向刚创建的索引添加合并后的mapping，最后构建新的matadata，并将集群新的matadata发送给各个节点完成索引创建。
总结
索引创建的过程包括三步：更新集群matadata，调用indiceService中创建索引，向新创建的索引中放置（合并到Index对应的mappingService中）mapping。这三步都在以上的两个方法中。完成这三步，集群中就保存了新索引的信息，同时索引配置和mapping放置也完成。索引就可以正常使用。
以上就是elasticsearch索引创建create index集群matedata更新的详细内容，更多关于elasticsearch索引create index matedata更新的资料请关注服务器之家其它相关文章！
原文链接：https://www.cnblogs.com/zziawanblog/p/6971379.html
