用PySpark索引从HDFS到弹性搜索的数据

Question

我正在尝试用Pyspark将数据索引到HDFS中的弹性搜索。当涉及到超过50 mb和不足100 mb的数据时，这些工作正在失败。

有人能告诉我如何优化这个过程吗？我有四个节点ES集群和相同的火花，每个有4g堆。

例外情况提及如下：

Caused by: org.apache.spark.util.TaskCompletionListenerException: Could not write all entries for bulk operation [1/1]. Error sample (first [5] error messages):
        org.elasticsearch.hadoop.rest.EsHadoopRemoteException: mapper_parsing_exception: failed to parse;org.elasticsearch.hadoop.rest.EsHadoopRemoteException: not_x_content_exception: not_x_content_exception: Compressor detection can only be called on some xcontent bytes or compressed xcontent bytes
        {"index":{}}

我使用的脚本如下：

df = spark.read.csv("hdfs://xxxxxx:9000/SOMEPATH",mode="DROPMALFORMED", header=True)
df_1.write.format("org.elasticsearch.spark.sql").option('es.nodes', 'xxxxxx').option('es.port',9200).option("es.node","lb")\
    .option("es.net.http.auth.user","xx").option("es.net.http.auth.pass","xx").option("es.input.json", "true")\
    .option('es.batch.write.retry.wait','60s').option('es.batch.size.entries','100').option('es.batch.write.retry.count' ,'6').option('es.resource','xx/xy').save()

注:我已经经历了可能的解决方案，但在我的情况下没有任何效果。

谢谢

Answer 1

问题是火花公子的罐子。如果我们使用Scala，它可以正常工作。因此，将我的管道代码更改为scala，它运行得非常完美。

Answer 2

当我删除代码：.option("es.input.json", true)时，它可以工作。