在最关键的条件下加入Spark DataFrames

Question

在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']
)

我们希望加入其中以产生以下结果DataFrame：

+--------+-----+------+-----------------------+
|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增加计算量，减少会降低结果质量。

两种方法似乎都不随输入DataFrames的大小线性缩放。

在实践中，我必须处理包含数千万行的输入数据，因此，我现在迷失了一个可行的解决方案。

FBJ方法

FBJ方法包括以下步骤：

1. 创建一个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)
)

2. 将ts_bin列上的DataFrames连接起来，通过以下方式实现：

ddf = ddf_event.join(ddf_gps, 'ts_bin', 'left_outer')

3. 确定最小时间戳差异，方法是：

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)
)

4. 过滤并选择相关的行和列，通过以下方式实现：

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'
    )
)

极限bin_size情况：

• bin_size >> 1 有效地导致完全交叉连接
• bin_size = 1 有效地导致左连接 ts_event == ts_gps

FSU方法

FSU方法包括以下步骤：

1. 联合数据框架，通过以下方式实现：

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)

2. 排序所得的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)
    )
)

3. 过滤并选择相关的行和列，通过以下方式实现：

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'
    )
)

Answer 1

您正在寻找的是时间连接。查看时间序列 Spark 库 Flint（以前称为 HuoHua，中文为 Spark）： https://github.com/twosigma/flint

使用此库，对于 2 个给定的时间序列数据帧（文档解释了这些对象），您可以在 PySpark（或 Scala Spark）中执行：

ddf_event = ...
ddf_gps = ...
result = ddf_event.leftJoin(ddf_gps, tolerance = "1day")

您的时间戳不清楚，因此请根据您的需要设置容差。如果需要，您还可以进行“未来连接”。

查看他们的 Spark Summit 演示，了解更多说明和示例： https://youtu.be/g8o5-2lLcvQ