Spark 数据帧加入范围慢

Question

我有以下用于 Spark 作业的输入数据（在 Parquet 中）：

Person (millions of rows)
+---------+----------+---------------+---------------+
|  name   | location |     start     |      end      |
+---------+----------+---------------+---------------+
| Person1 |     1230 | 1478630000001 | 1478630000010 |
| Person2 |     1230 | 1478630000002 | 1478630000012 |
| Person2 |     1230 | 1478630000013 | 1478630000020 |
| Person3 |     3450 | 1478630000001 | 1478630000015 |
+---------+----------+---------------+---------------+


Event (millions of rows)
+----------+----------+---------------+
|  event   | location |  start_time   |
+----------+----------+---------------+
| Biking   |     1230 | 1478630000005 |
| Skating  |     1230 | 1478630000014 |
| Baseball |     3450 | 1478630000015 |
+----------+----------+---------------+

我需要将其转换为以下预期结果：

[{
    "name" : "Biking",
    "persons" : ["Person1", "Person2"]
},
{
    "name" : "Skating",
    "persons" : ["Person2"]
},
{
    "name" : "Baseball",
    "persons" : ["Person3"]
}]

换句话说：结果是每个事件的列表，每个事件都有参与此事件的人员列表。

如果

一个人算作参与者

Person.start < Event.start_time 
&& Person.end > Event.start_time
&& Person.location == Event.location

我尝试过不同的方法，但唯一有效的方法是加入两个数据帧，然后按事件 group/aggregate 它们。但是连接速度非常慢，并且不能很好地分布在多个 CPU 核心上。

加入的当前代码：

final DataFrame fullFrame = persons.as("persons")
    .join(events.as("events"), col("persons.location").equalTo(col("events.location"))
               .and(col("events.start_time").geq(col("persons.start")))
               .and(col("events.start_time").leq(col("persons.end"))), "inner");

//count to have an action 
fullFrame.count();

我正在使用 Spark Standalone 和 Java，如果这有区别的话。

有没有人知道如何使用 Spark 1.6.2 解决这个问题？

Answer 1

范围连接作为与后续过滤步骤的叉积执行。一个可能更好的解决方案可能是 broadcast 可能更小的 events table 然后映射 persons table: 在地图里面, 检查连接条件并产生相应的结果。

Spark 数据帧加入范围慢

Spark dataframe join with range slow

java

apache-spark

apache-spark-sql

spark-dataframe