从spark写入elasticsearch非常慢答案

【问题标题】：Write to elasticsearch from spark is very slow从spark写入elasticsearch非常慢
【发布时间】：2018-05-07 06:52:32
【问题描述】：

我正在处理一个文本文件并将转换后的行从 Spark 应用程序写入弹性搜索，如下所示

input.write.format("org.elasticsearch.spark.sql")
      .mode(SaveMode.Append)
      .option("es.resource", "{date}/" + dir).save()

这运行速度非常慢，写入 287.9 MB / 1513789 条记录大约需要 8 分钟。

鉴于始终存在网络延迟，我如何调整 spark 和 elasticsearch 设置以使其更快。

我在本地模式下使用 spark，有 16 个内核和 64GB RAM。我的 elasticsearch 集群有 1 个主节点和 3 个数据节点，每个节点有 16 个内核和 64GB。

我正在阅读如下文本文件

 val readOptions: Map[String, String] = Map("ignoreLeadingWhiteSpace" -> "true",
  "ignoreTrailingWhiteSpace" -> "true",
  "inferSchema" -> "false",
  "header" -> "false",
  "delimiter" -> "\t",
  "comment" -> "#",
  "mode" -> "PERMISSIVE")

....

val input = sqlContext.read.options(readOptions).csv(inputFile.getAbsolutePath)

【问题讨论】：

输入中有多少个分区？ spark 是否与 elasticsearch 共享资源？您可以随时更改默认写入批量大小，我相信默认为 500。
实际上我正在使用 csv 插件 val input = sqlContext.read.options(readOptions).csv(inputFile.getAbsolutePath) 创建数据框。我不知道它创建了多少个分区。 Spark 不与 elasticsearch 共享资源。
你的文件有多大？它是分区的还是一个大集团？
它是 70MB gz 文件，包含网络访问日志。这是一个未分区的文件
我想你在你的 readOptions 中推断Schema，这将导致扫描数据两次。你能打印 df.rdd.getNumPartitions 的输出吗？

标签： apache-spark elasticsearch elasticsearch-5 elasticsearch-spark

【解决方案1】：

首先，让我们从您的应用程序中发生的事情开始。 Apache Spark 正在读取 1 个（不是那么大）csv 压缩文件。因此，第一个 spark 将花时间解压缩数据并在将其写入 elasticsearch 之前对其进行扫描。

这将创建一个Dataset/DataFrame 一个分区（由您在 cmets 中提到的df.rdd.getNumPartitions 的结果确认）。

一个直接的解决方案是在将数据写入elasticsearch 之前读取并缓存您的数据repartition。现在我不确定你的数据是什么样的，所以决定分区的数量是你这边基准的主题。

val input = sqlContext.read.options(readOptions)
                      .csv(inputFile.getAbsolutePath)
                      .repartition(100) // 100 is just an example
                      .cache

我不确定对您的应用程序有多大好处，因为我相信可能存在其他瓶颈（网络 IO、ES 的磁盘类型）。

PS：在构建 ETL 之前，我应该将 csv 转换为 parquet 文件。这里有真正的性能增益。（个人意见和基准）

另一个可能的优化是调整 elasticsearch-spark 连接器的 es.batch.size.entries 设置。默认值为1000。

设置此参数时需要小心，因为您可能会使 elasticsearch 过载。我强烈建议您查看可用的配置here。

我希望这会有所帮助！

【讨论】：