如何在不查询每个节点的情况下使用 PySpark 对 Elasticsearch 运行查询？答案

【问题标题】：How do I run a query against Elasticsearch using PySpark without querying every node?如何在不查询每个节点的情况下使用 PySpark 对 Elasticsearch 运行查询？
【发布时间】：2019-07-06 17:25:52
【问题描述】：

我的最终目标是使用 PySpark 高效地索引 Elasticsearch (ES) 中的大量数据，然后针对索引运行大量查询并记录结果的统计信息。

Elasticsearch version 5.6.5
Spark version 2.4.0
Hadoop version 2.7
Elasticsearch-Hadoop python library version: 6.6.0

考虑以下代码：

from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SQLContext

# create our Spark Context  
sc_conf = SparkConf().setAll((
    ("spark.serializer", "org.apache.spark.serializer.KryoSerializer"),
))
sc_conf.setAppName("PythonSparkStreaming")

sc = SparkContext(conf=sc_conf)

sqlContext = SQLContext(sc)

q ="""{
  "query": {
    "match_all": {}
  }  
}"""

es_live_conf["es.query"] = q

es_rdd = sc.newAPIHadoopRDD(
    inputFormatClass="org.elasticsearch.hadoop.mr.EsInputFormat",
    keyClass="org.apache.hadoop.io.NullWritable", 
    valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable", 
    conf=es_live_conf)

sqlContext.createDataFrame(es_rdd).limit(1).collect()

我只是想对索引运行 match all 查询，我只想要最高的结果。我尝试在 ES 查询中表达限制，但显然 Spark 忽略了这一点，因此我使用数据框过滤器来表达它。

我的 Spark 配置如下：

es_live_conf = {

# specify the node that we are sending data to (this should be the master)
"es.nodes" : 'xxx.xxxx.com',

# specify the port in case it is not the default port
"es.port" : ES_PORT,

# specify a resource in the form 'index/doc-type'
"es.resource" : 'xxxxxx/document',

"es.net.http.auth.user" : ES_USERNAME,

"es.net.http.auth.pass" : ES_PASSWORD,

"es.net.ssl":"true",

"es.net.ssl.cert.allow.self.signed": "true",

"es.nodes.discovery": "false",

"es.nodes.wan.only": "true",

"es.index.read.missing.as.empty": "true",

}

我正在访问 VPC 后面的 ES 集群，因此我只能访问客户端节点，而不能访问内部数据等节点。这就是 wan.only 设置为 true 的原因。

通过这种设置，Spark 似乎会以完全匹配的方式查询每个节点，然后最终合并为我真正想要的单个结果。它非常慢（50 个分片，3000 万个文档），它完全避免了 ES 有效减少每个节点本身的结果的能力。即使我将查询更改为通过单个文档 ID 专门搜索，它也会通过主节点对每个单独的分片运行查询，方法是在每次调用时指定特定的分片 ID。我尝试将es.nodes.client.only 设置为true，但这抱怨设置与wan.only 冲突。如果我启用client.only 并禁用wan.only，我将无法再连接到集群，因为它会尝试直接连接无法访问的每个节点。

我在这里做错了什么？如何使用 PySpark 对 ES 运行一次查询，而不是为每个分片运行一次。此外，如果 PySpark 尝试在每个分片上运行完整查询然后看似 post 处理结果，我如何在我的查询中使用 from、size 和 rescore 之类的东西？

【问题讨论】：

标签： python apache-spark elasticsearch pyspark elasticsearch-hadoop

【解决方案1】：

我找不到使用 ES Hadoop 库解决此问题的方法。当您需要对单个 Elasticsearch 查询返回的结果运行非常长、非常复杂的 reduce 步骤时，它似乎更适合使用 Spark，而不是运行数百万个快速 ES 查询并聚合结果。为了解决这个问题，我最终使用了这个插件：https://github.com/sourav-mazumder/Data-Science-Extensions/tree/master/spark-datasource-rest

我实际上进一步开发了它，以便每个内核可以使用多个线程来并行执行更多请求。它不仅可以让您对 ES 集群进行 DDOS，还可以为您可能需要消耗和聚合大量请求的任何平台提供任何静态端点。

如果我能搞清楚，我也会在 github 上发布我公开创建的多线程版本。

【讨论】：