调优弹性写入性能

Question

我试图在Elasticsearch集群中保存(和索引)170 in文件(大约9.15亿行和25列)。我在一个5节点的elasticsearch集群上获得了糟糕的性能。这项任务花费了5小时。星火集群有150个核心10倍(15 CPU，64 RAM)。

这是我目前的担忧：

• 从S3的多个拼花文件构建一个Spark。
• 然后使用Spark的"org.elasticsearch.spark.sql“源将该数据保存到ElasticSearch索引中。(我尝试了许多performance)

切分和复制配置组合，但没有获得。

--这是集群节点的特征

each.

• HEAP_SIZE是
• 5节点(16 CPU，64 RAM，700 on磁盘)的50%，即每个节点上的32 on。在/etc/elasticsearch/jvm.options

中配置

这是将数据文件写入ElasticSearch的代码(用scala编写)

 writeDFToEs(whole_df, "main-index")

写function函数：

  def writeDFToEs(df: DataFrame, index: String) = {
    df.write
      .format("org.elasticsearch.spark.sql")
      .option("es.nodes", "192.168.1.xxx")
      .option("es.http.timeout", 600000)
      .option("es.http.max_content_length", "2000mb")
      .option("es.port", 9200)
      .mode("overwrite")
      .save(s"$index")

  }

你能帮我找出我做得不好的地方吗？

提前谢谢。

Answer 1

回答了我自己的问题.

正如@warkolm所建议的，我专注于_bulk。

我使用的是es-hadoop连接器，所以我不得不调整es.batch.size.entries参数。

在运行了一系列测试(测试各种值)之后，我最终获得了更好的结果(尽管不是最优的)，es.batch.size.entries设置为10000，ES索引模板中的值如下。

{
  "index": {
    "number_of_shards": "10",
    "number_of_replicas": "0",
    "refresh_interval": "60s"
  }
}

最后，我的df.write看起来如下：

    df.write
      .format("org.elasticsearch.spark.sql")
      .option("es.nodes", es_nodes)
      .option("es.port", es_port)
      .option("es.http.timeout", 600000)
      .option("es.batch.size.entries", 10000)
      .option("es.http.max_content_length", "2000mb")
      .mode("overwrite")
      .save(s"$writeTo")

现在这个过程需要3h (2h 55分钟)，而不是5个小时。

我仍然在改进信任和代码。如果我有更好的表现我会更新的。