Spark向Elasticsearch批量导入数据,出现重复记录问题的定位
发表于:2026-11-01 作者:千家信息网编辑
千家信息网最后更新 2026年11月01日,看了下es-hadoop插件的源码:发现ES导入数据重试情况的发生,除了在es.batch.write.retry.policy参数默认开启且es-hadoop插件向ES集群发送bulk写入请求接受到
千家信息网最后更新 2026年11月01日Spark向Elasticsearch批量导入数据,出现重复记录问题的定位
看了下es-hadoop插件的源码:
发现ES导入数据重试情况的发生,除了在es.batch.write.retry.policy参数默认开启且es-hadoop插件向ES集群发送bulk写入请求接受到503响应码会重试3次之外。
本身执行http请求时,也会存在重试(hadoop/rest/NetworkClient.java):
public Response execute(Request request) { Response response = null; boolean newNode; do { SimpleRequest routedRequest = new SimpleRequest(request.method(), null, request.path(), request.params(), request.body()); newNode = false; try { response = currentTransport.execute(routedRequest); ByteSequence body = routedRequest.body(); if (body != null) { stats.bytesSent += body.length(); } } catch (Exception ex) { // configuration error - including SSL/PKI - bail out if (ex instanceof EsHadoopIllegalStateException) { throw (EsHadoopException) ex; } // issues with the SSL handshake, bail out instead of retry, for security reasons if (ex instanceof javax.net.ssl.SSLException) { throw new EsHadoopTransportException(ex); } // check for fatal, non-recoverable network exceptions if (ex instanceof BindException) { throw new EsHadoopTransportException(ex); } if (log.isTraceEnabled()) { log.trace( String.format( "Caught exception while performing request [%s][%s] - falling back to the next node in line...", currentNode, request.path()), ex); } String failed = currentNode; failedNodes.put(failed, ex); newNode = selectNextNode(); log.error(String.format("Node [%s] failed (%s); " + (newNode ? "selected next node [" + currentNode + "]" : "no other nodes left - aborting..."), failed, ex.getMessage())); if (!newNode) { throw new EsHadoopNoNodesLeftException(failedNodes); } } } while (newNode); return response; }当请求出现超时的情况时,es-hadoop插件会再请求一个ES节点发送写入请求。即导入插件认为当前插入节点超时了(默认是一分钟)就视为该节点不可用,就换下一个节点,其实是ES在一分钟内没有处理完插入任务。
将超时时间es.http.timeout参数调大之后,给ES留下充足的入库时间,就不会再发生这个问题了。
插件
节点
参数
情况
时间
数据
问题
充足
任务
源码
集群
处理
定位
数据库的安全要保护哪些东西
数据库安全各自的含义是什么
生产安全数据库录入
数据库的安全性及管理
数据库安全策略包含哪些
海淀数据库安全审计系统
建立农村房屋安全信息数据库
易用的数据库客户端支持安全管理
连接数据库失败ssl安全错误
数据库的锁怎样保障安全
南京欣网互联网络科技
互联网科技网
鑫亿盛互联网科技
天下捷融互联网科技
互联网科技网站
亿人帮互联网科技
互联网科技宣传片
中国最先进互联网科技公司
互联网科技产品
中国科技互联网公司收入
互联网经济与金融科技
港股科技和互联网
玉溪互联网科技怎么选
云南省自由互联网科技有限公司
互联网与数字科技关系
上海同邻互联网科技有限公司规模
科技互联网经济论坛
保山浩联互联网科技有限公司
网互联网络科技
前海云享互联网科技
互联网科技名片
涨乐互联网科技
互联网品科技
南京欣网互联网科技
哈尔滨新华互联网科技
四个字互联网科技公司取名
博兴的互联网电子科技公司
新兴科技互联网创新大赛颁奖典礼
速维有品互联网信息科技(苏州)有限公司
陕西彩京云互联网科技有限公司