将 rdd 从 spark 写入 Elastic Search 失败答案

【问题标题】：writing rdd from spark to Elastic Search fails将 rdd 从 spark 写入 Elastic Search 失败
【发布时间】：2017-03-26 01:13:12
【问题描述】：

我正在尝试在版本 2.4.0 上为 Elastic Cloud 上的 Elastic Search 编写一对 rdd。我正在使用elasticsearch-spark_2.10-2.4.0 插件写入 ES。这是我用来写入 ES 的代码：

def predict_imgs(r):  
  import json
  out_d = {}
  out_d["pid"] = r["pid"]
  out_d["other_stuff"] = r["other_stuff"]

  return (r["pid"], json.dumps(out_d))

res2 = res1.map(predict_imgs)

es_write_conf = {
"es.nodes" : image_es,
#"es.port" : "9243",
"es.resource" : "index/type",
"es.nodes.wan.only":"True",
"es.write.operation":"upsert",
"es.mapping.id":"product_id",
"es.nodes.discovery" : "false",
"es.net.http.auth.user": "username",
"es.net.http.auth.pass": "pass",
"es.input.json": "true",
"es.http.timeout":"1m",
"es.scroll.size":"10",
"es.batch.size.bytes":"1mb",
"es.http.retries":"1",
"es.batch.size.entries":"5",
"es.batch.write.refresh":"False",
"es.batch.write.retry.count":"1",
"es.batch.write.retry.wait":"10s"}

res2.saveAsNewAPIHadoopFile(
path='-', 
outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat",
keyClass="org.apache.hadoop.io.NullWritable", 
valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable", 
conf=es_write_conf)

我得到的错误如下：

Py4JJavaError: An error occurred while calling     z:org.apache.spark.api.python.PythonRDD.saveAsNewAPIHadoopFile.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 744 in stage 26.0 failed 4 times, most recent failure: Lost task 744.3 in stage 26.0 (TID 2841, 10.181.252.29): org.apache.spark.SparkException: Python worker exited unexpectedly (crashed)

有趣的是，当我对 rdd2 上的前几个元素进行处理，然后从中创建一个新的 rdd 并将其写入 ES 时，它可以正常工作：

x = sc.parallelize([res2.take(1)])
x.saveAsNewAPIHadoopFile(
path='-', 
outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat",
keyClass="org.apache.hadoop.io.NullWritable", 
valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable", 
conf=es_write_conf)

我正在使用 Elastic Cloud（Elastic Search 的云产品）和 Databricks（Apache Spark 的云产品）难道是 ES 无法跟上 Spark 写入 ES 的吞吐量？我将 Elastic Cloud 的大小从 2GB RAM 增加到 8GB RAM。

我上面使用的es_write_conf 有什么推荐的配置吗？您还能想到其他任何confs 吗？更新到 ES 5.0 有帮助吗？

感谢任何帮助。这几天一直在为此苦苦挣扎。谢谢。

【问题讨论】：

标签： hadoop elasticsearch apache-spark databricks

【解决方案1】：

这看起来像是 pyspark 计算的问题，而不是必要的弹性搜索保存过程。通过以下方式确保您的 RDD 正常：

在 rdd1 上执行 count()（以“实现”结果）
在 rdd2 上执行 count()

如果计数正常，请在保存到 ES 之前尝试缓存结果：

res2.cache()
res2.count() # to fill the cache
res2.saveAsNewAPIHadoopFile(...

如果问题仍然存在，请尝试查看已死的执行器 stderr 和 stdout（您可以在 SparkUI 的 Executors 选项卡上找到它们）。

我还注意到es_write_conf 中的批处理大小非常小，尝试将其增加到 500 或 1000 以获得更好的性能。

【讨论】：