elasticsearch - Spark : optimized join with ElasticSearch index

Question

所以我正在学习通过 Apache Spark 从 ElasticSearch 获取数据。假设我已连接到具有“用户”索引的 ElasticSearch。

sqlContext = SQLContext(sc)
usersES=sqlContext.read.format('org.elasticsearch.spark.sql').option('es.nodes','mynode').load('users/user')

explain(usersES) 向我展示了这个:

== Physical Plan ==

Scan ElasticsearchRelation(Map(es.nodes -> mynode, es.resource -> users/user),org.apache.spark.sql.SQLContext@6c78e806,None)[about#145,activities#146,bdate#147, uid#148]

当我使用过滤器时:

usersES.filter(usersES.uid==1566324).explain()

== Physical Plan == Filter (uid#203L = 1566324) +- Scan ElasticsearchRelation(Map(es.nodes -> mynode, es.resource -> users/user),org.apache.spark.sql.SQLContext@6c78e806,None)[about#145,activities#146,bdate#147,uid#148] PushedFilters: [EqualTo(uid,1566324)]

如您所见，Spark 优雅地将过滤器推送到 ElasticSearch，使索引搜索变得快速和舒适。

但是当我尝试将 userES 与另一个数据框连接时，我总是遇到同样的问题:Spark 扫描整个 ElasticSearch 索引，不推送我给它的任何过滤器。例如:

a = sc.parallelize([1566324,1566329]).map(Row('id')).toDF()
a.join(usersES, usersES.uid==a.id).explain()

显示:

SortMergeJoin [id#210L], [uid#203L] :- Sort [id#210L ASC], false, 0 : +- TungstenExchange hashpartitioning(id#210L,200), None : +- ConvertToUnsafe : +- Scan ExistingRDD[id#210L] +- Sort [uid#203L ASC], false, 0 +- TungstenExchange hashpartitioning(uid#203L,200), None +- ConvertToUnsafe +- Scan ElasticsearchRelation(Map(es.nodes -> mynode, es.resource -> users/user),org.apache.spark.sql.SQLContext@6c78e806,None)[about#145,activities#146,bdate#147,uid#148]

请告诉我，是否可以在连接内的 Elasticsearch 中推送过滤器？

Answer 1

这是预期的行为，是的，elaticsearch-hadoop 连接器支持下推谓词，但加入时没有推送。

这是因为连接操作不知道键在数据帧中是如何分区的。

默认情况下，此操作将散列两个数据帧的所有键，通过网络将具有相同键散列的所有元素发送到同一台机器，然后在该机器上将具有相同键的元素连接在一起。

这就是为什么您在没有下推谓词的情况下获得执行计划的原因。

编辑: 从 2.1 版开始，连接器似乎支持 IN 子句。如果您的 DataFrame a 不大，您应该使用它。

• 引用 1。https://github.com/elastic/elasticsearch-hadoop/issues/556
• 引用文献 2。https://github.com/elastic/elasticsearch-hadoop/commit/bbe6154226f4700a9ac23c3d611b6c313bfbaf59