nullpointerexception在使用带有spark流的elastic search 5.5批量摄取api时

j1dl9f46 于 2021-06-07 发布在 Kafka

关注(0)|答案(2)|浏览(289)

获取nullpointerexception:

java.lang.NullPointerException
    at org.elasticsearch.action.bulk.BulkRequest.validate(BulkRequest.java:604)
    at org.elasticsearch.action.TransportActionNodeProxy.execute(TransportActionNodeProxy.java:46)
    at org.elasticsearch.client.transport.TransportProxyClient.lambda$execute$0(TransportProxyClient.java:59)
    at org.elasticsearch.client.transport.TransportClientNodesService.execute(TransportClientNodesService.java:250)
    at org.elasticsearch.client.transport.TransportProxyClient.execute(TransportProxyClient.java:59)
    at org.elasticsearch.client.transport.TransportClient.doExecute(TransportClient.java:363)
    at org.elasticsearch.client.support.AbstractClient.execute(AbstractClient.java:408)
    at org.elasticsearch.action.ActionRequestBuilder.execute(ActionRequestBuilder.java:80)
    at org.elasticsearch.action.ActionRequestBuilder.execute(ActionRequestBuilder.java:54)

我有一个场景，多个并发任务在spark streaming应用程序中的4个执行器中运行，每个执行器从kafka读取数据，准备批量并接收es索引中的一批记录。我第一次在这些批记录中得到了一个奇怪的nullpointerexception，但是在第二次运行中它们得到了成功的处理。
有人能告诉我为什么会这样吗。

scala elasticsearch apache-kafka apache-spark

来源：https://stackoverflow.com/questions/48781385/nullpointerexception-while-using-elastic-search-5-5-bulk-ingest-api-with-spark-s

2条答案

按热度按时间

8xiog9wr1#

这是我正在使用的代码段，第一行是build.sbt文件中的依赖项

//lib dependency in build.sbt
"org.elasticsearch" %% "elasticsearch-spark-20" % "5.6.5"

//below is the connection variables required by Spark

val resources: String =
  s"${appConf.getString("es-index")}/${appConf.getString("es.type")}"
val esConfig: Map[String, String] = Map(
  "es.index.auto.create" -> s"${appConf.getString("es.index.auto.create")}",
  "es.nodes" -> s"${appConf.getString("es-nodes")}",
  "es.port" -> s"${appConf.getInt("es.port")}",
  "es.nodes.wan.only" -> s"${appConf.getString("es.nodes.wan.only")}",
  "es.net.ssl" -> s"${appConf.getString("es.net.ssl")}"
)

import org.elasticsearch.spark._
    val dstream: InputDStream[ConsumerRecord[String, String]] =
  KafkaUtils.createDirectStream[String, String](
    ssc,
    LocationStrategies.PreferConsistent,
    ConsumerStrategies.Subscribe[String, String](conn.topic,
                                                 conn.kafkaProps)
  )
dstream.foreachRDD(rdd =>
  rdd.map(_.value).saveJsonToEs(resources,esConfig))
ssc.checkpoint("/tmp/OACSpark")
ssc.start()
ssc.awaitTermination()

我使用typesafe config从属性文件读取config。我以json的形式将数据发布到kafka，因此我使用了“savejsontoes（）”api，您可以在elasticsearch网站上的连接器文档中找到更多信息

赞(0）回复(0）举报 2021-06-07

mtb9vblg2#

到目前为止，我找到了一个解决方法，可以一次将记录推送到es索引，并删除了这个bulkapi（bulkapi在后台也做同样的事情）。

赞(0）回复(0）举报 2021-06-07