在 PySpark 中执行模糊连接的高效方法是什么？
我正在寻找社区对在最近的关键条件下加入大型 Spark DataFrame 的可扩展方法的看法。请允许我通过一个有代表性的例子来说明这个问题。假设我们有以下 Spark DataFrame，其中包含在某个时间点发生的事件：
ddf_event = spark.createDataFrame(
data=[
[1, 'A'],
[5, 'A'],
[10, 'B'],
[15, 'A'],
[20, 'B'],
[25, 'B'],
[30, 'A']
],
schema=['ts_event', 'event']
)
以及以下包含在某个时间点测量的 GPS 数据的 Spark DataFrame：
ddf_gps = spark.createDataFrame(
data=[
[2, '(-46.84635, 173.13674)'],
[4, '(2.50362, 104.34136)'],
[8, '(-24.20741, 51.80755)'],
[15, '(-59.07798, -20.49141)'],
[18, '(-44.34468, -167.90401)'],
[24, '(-18.84175, 16.68628)'],
[27, '(20.48501,58.42423)']
],
schema=['ts_gps', 'gps_coordinates']
)
我们希望加入它以生成以下结果数据帧：
+--------+-----+------+-----------------------+
|ts_event|event|ts_gps|gps_coordinates |
+--------+-----+------+-----------------------+
|1 |A |2 |(-46.84635, 173.13674) |
|5 |A |4 |(2.50362, 104.34136) |
|10 |B |8 |(-24.20741, 51.80755) |
|15 |A |15 |(-59.07798, -20.49141) |
|20 |B |18 |(-44.34468, -167.90401)|
|25 |B |24 |(-18.84175, 16.68628) |
|30 |A |27 |(20.48501,58.42423) |
+--------+-----+------+-----------------------+
给定事件时间戳和 GPS 数据时间戳，有效找到最近的 GPS 数据点。
因此，我们遇到了连接最近关键条件的问题，在这种情况下，“最近”被定义为时间戳之间的最小绝对差。
我探索了两种方法来实现此目的：一种基于过滤分箱连接 (FBJ)，一种基于过滤排序联合 (FSU)。下面更详细地描述这两种方法。
FBJ 方法取决于参数bin_size，这限制了可以找到匹配 GPS 时间戳的时间窗口。增加bin_size增加计算负载，减少计算负载会降低结果质量。
这两种方法似乎都不随输入 DataFrame 的大小线性缩放。
在实践中，我必须处理由数千万行组成的输入数据，因此我目前找不到解决该问题的可行解决方案。
FBJ方法
FBJ 方法包括以下步骤：
- 创建一个
ts_bin列，将timestamp列，实施者：
bin_size = 10
ddf_event = ddf_event.withColumn(
'ts_bin',
F.round(F.col('ts_event') / bin_size)
)
ddf_gps = ddf_gps.withColumn(
'ts_bin',
F.round(F.col('ts_gps') / bin_size)
)
- 加入 DataFrames
ts_bin专栏，实施者：
ddf = ddf_event.join(ddf_gps, 'ts_bin', 'left_outer')
- 确定最小时间戳差异，实现方式为：
from pyspark.sql.window import Window
window = Window.partitionBy('ts_event')
ddf = ddf.withColumn(
'ts_diff',
F.abs(F.col('ts_gps') - F.col('ts_event'))
)
ddf = ddf.withColumn(
'min_ts_diff',
F.min(F.col('ts_diff')).over(window)
)
- 过滤并选择相关的行和列，实现方式为：
ddf = (
ddf
.where(
(F.col('ts_diff') == F.col('min_ts_diff')) |
(F.col('ts_diff').isNull())
)
.select(
'ts_event',
'event',
'ts_gps',
'gps_coordinates'
)
)
Limit bin_size情况：
-
bin_size >> 1有效地产生完全交叉连接
-
bin_size = 1有效地导致左连接ts_event == ts_gps
前苏联方法
FSU 方法包括以下步骤：
- 联合 DataFrames，实现者：
def union(df1, df2):
cols = list(set(df1.columns).union(set(df2.columns)))
for col in cols:
if col not in df1.columns:
df1 = df1.withColumn(col, F.lit(None))
if col not in df2.columns:
df2 = df2.withColumn(col, F.lit(None))
return df1.select(cols).union(df2.select(cols))
ddf_event = ddf_event.withColumn('timestamp', F.col('ts_event'))
ddf_gps = ddf_gps.withColumn('timestamp', F.col('ts_gps'))
ddf = union(ddf_event, ddf_gps)
- 对生成的 DataFrame 进行排序并获取相邻的 GPS 时间戳，实现方式为：
from sys import maxsize
last_window = Window.orderBy(
F.col('timestamp').asc()).rowsBetween(-maxsize, 0)
first_window = Window.orderBy(
F.col('timestamp').asc()).rowsBetween(0, maxsize)
ddf = (
ddf.withColumn(
'prev_time',
F.last(F.col('ts_gps'), ignorenulls=True)
.over(last_window)
).withColumn(
'prev_coordinates',
F.last(F.col('gps_coordinates'), ignorenulls=True)
.over(last_window)
).withColumn(
'next_time',
F.first(F.col('ts_gps'), ignorenulls=True)
.over(first_window)
).withColumn(
'next_coordinates',
F.first(F.col('gps_coordinates'), ignorenulls=True)
.over(first_window)
)
)
- 过滤并选择相关的行和列，实现方式为：
condition = (F.col('timestamp') - F.col('prev_time')
< F.col('next_time') - F.col('timestamp'))
ddf = (
ddf
.where(F.col('event').isNotNull())
.withColumn(
'ts_gps',
F.when(condition | F.col('next_time').isNull(), F.col('prev_time')).otherwise(F.col('next_time'))
).withColumn(
'gps_coordinates',
F.when(condition | F.col('next_time').isNull(),
F.col('prev_coordinates'))
.otherwise(F.col('next_coordinates'))
).select(
'ts_event',
'event',
'ts_gps',
'gps_coordinates'
)
)
