3

我正在尝试在版本上为 Elastic Cloud 上的 Elastic Search 编写一对 rdd 2.4.0。我正在使用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 有帮助吗？

任何帮助表示赞赏。这几天一直在为此苦苦挣扎。谢谢你。

4

1 回答 1

2

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

  1. 在rdd1count()上执行（以“实现”结果）
  2. count()在 rdd2 上执行

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

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

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

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

于 2016-11-11T20:34:24.403 回答