【发布时间】：2016-08-23 18:33:16
【问题描述】：
我在 azure 上有 3 个主节点 + 3 个数据节点弹性搜索集群。我正在尝试执行批量操作，但我收到有关节点本身的失败错误，这是我设置客户端的方式：
final Builder builder = Settings.builder();
final org.elasticsearch.client.transport.TransportClient.Builder transBuilder = TransportClient.builder();
builder.put("cluster.name", esCluster);
if (esShield) {
builder.put("shield.user", esUsername + ":" + esPassword);
transBuilder.addPlugin(ShieldPlugin.class);
}
final Settings settings = builder.build();
TransportClient esClient = transBuilder.settings(settings).build();
final String[] hosts = esHost.split(",");
for (String host : hosts) {
esClient.addTransportAddress(new InetSocketTransportAddress(new InetSocketAddress(host, Integer.parseInt(esPort))));
}
这里是批量操作：
BulkProcessor bulkProcessor = BulkProcessor.builder(getClient(), new BulkProcessor.Listener() {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
LOGGER.info("Going to execute new bulk composed of {" + request.numberOfActions() + "} actions");
}
@Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
LOGGER.info("Executed bulk composed of {" + request.numberOfActions() + "} actions");
}
@Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
LOGGER.info("Error executing bulk");
failure.printStackTrace();
}
}).setBulkActions(docs.size()).setConcurrentRequests(250).build();
for (DBObject doc : docs) {
bulkProcessor.add(getClient().prepareIndex(indexName, typeName).setSource(doc.toMap()).request());
}
它开始对 1,000 个这样的记录批次做出良好响应：
将执行由 {1001} 个操作组成的新批量
由 {1001} 个操作组成的批量执行
然后我开始收到以下错误：
transport:383 - [Stanley Stewart] 无法获取 {#transport#-1}{10.0.0.10}{10.0.0.10:9300} 的节点信息，正在断开连接... ReceiveTimeoutTransportException[[][10.0.0.10:9300][cluster:monitor/nodes/liveness] request_id [60] 在 [5000ms] 后超时 在 org.elasticsearch.transport.TransportService$TimeoutHandler.run(TransportService.java:679) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:745)
最后我得到以下错误：
bulk:148 - [Stanley Stewart] 未能执行批量请求 1。 NoNodeAvailableException[没有配置的节点可用：[{#transport#-1}{10.0.0.10}{10.0.0.10:9300}，{#transport#-2}{10.0.0.11}{10.0.0.11:9300} , {#transport#-3}{10.0.0.12}{10.0.0.12:9300}]] 在 org.elasticsearch.client.transport.TransportClientNodesService.ensureNodesAreAvailable(TransportClientNodesService.java:290) 在 org.elasticsearch.client.transport.TransportClientNodesService.execute(TransportClientNodesService.java:207) 在 org.elasticsearch.client.transport.support.TransportProxyClient.execute(TransportProxyClient.java:55) 在 org.elasticsearch.client.transport.TransportClient.doExecute(TransportClient.java:288) 在 org.elasticsearch.client.support.AbstractClient.execute(AbstractClient.java:359) 在 org.elasticsearch.client.support.AbstractClient.bulk(AbstractClient.java:436) 在 org.elasticsearch.action.bulk.Retry$AbstractRetryHandler.execute(Retry.java:219) 在 org.elasticsearch.action.bulk.Retry.withAsyncBackoff(Retry.java:72) 在 org.elasticsearch.action.bulk.BulkRequestHandler$AsyncBulkRequestHandler.execute(BulkRequestHandler.java:121) 在 org.elasticsearch.action.bulk.BulkProcessor.execute(BulkProcessor.java:312) 在 org.elasticsearch.action.bulk.BulkProcessor.executeIfNeeded(BulkProcessor.java:303) 在 org.elasticsearch.action.bulk.BulkProcessor.internalAdd(BulkProcessor.java:285) 在 org.elasticsearch.action.bulk.BulkProcessor.add(BulkProcessor.java:268) 在 org.elasticsearch.action.bulk.BulkProcessor.add(BulkProcessor.java:264) 在 org.elasticsearch.action.bulk.BulkProcessor.add(BulkProcessor.java:250)
谁能帮我弄清楚发生了什么以及如何解决它？
【问题讨论】：
-
根据我的经验，批量操作在 ES 中非常脆弱。我不了解 Azure，但通过我的本地安装，我为 ES 提供了尽可能多的内存。然后我尝试了单个散装的大小和散装之间的时间间隔。在您的情况下，尝试将批量降低到 100 个文档并引入时间间隔，例如 1 秒。尝试一下。
标签： java azure elasticsearch bulkinsert