我正在使用PySpark，并且即将放弃我的问题。我有两个数据集：一个非常非常大的数据集(集A(和一个相当小的数据集。它们的形式是：
Data set A:
variable | timestampA
---------------------------------
x | 2015-01-01 09:29:21
y | 2015-01-01 12:01:57
Data set B:
different information | timestampB
-------------------------------------------
info a | 2015-01-01 09:30:00
info b | 2015-01-01 09:30:00
info a | 2015-01-01 12:00:00
info b | 2015-01-01 12:00:00
A有许多行，每一行都有不同的时间戳。B每隔几分钟就有一个时间戳。这里的主要问题是，在两个数据集中都没有匹配的确切时间戳。
我的目标是在最近的时间戳上加入数据集。另外一个问题出现了，因为我想以特定的方式加入。对于A中的每个条目，我想在复制A中的条目的同时，将整个信息映射到最近的时间戳。因此，结果应该是：
Final data set
variable | timestampA | information | timestampB
--------------------------------------------------------------------------
x | 2015-01-01 09:29:21 | info a | 2015-01-01 09:30:00
x | 2015-01-01 09:29:21 | info b | 2015-01-01 09:30:00
y | 2015-01-01 12:01:57 | info a | 2015-01-01 12:00:00
y | 2015-01-01 12:01:57 | info b | 2015-01-01 12:00:00
我对PySpark(以及stackoverflow(非常陌生。我想我可能需要使用窗口功能和/或广播加入，但我真的没有必要开始，希望能得到任何帮助。非常感谢。
您可以使用broadcast来避免混洗。
如果理解正确，您在set_B中有时间戳，这些时间戳是某个确定的间隔的结果？如果是这样，您可以执行以下操作：
from pyspark.sql import functions as F
# assuming 5 minutes is your interval in set_B
interval = 'INTERVAL {} SECONDS'.format(5 * 60 / 2)
res = set_A.join(F.broadcast(set_B), (set_A['timestampA'] > (set_B['timestampB'] - F.expr(interval))) & (set_A['timestampA'] <= (set_B['timestampB'] + F.expr(interval))))
输出：
+--------+-------------------+------+-------------------+
|variable| timestampA| info| timestampB|
+--------+-------------------+------+-------------------+
| x|2015-01-01 09:29:21|info a|2015-01-01 09:30:00|
| x|2015-01-01 09:29:21|info b|2015-01-01 09:30:00|
| y|2015-01-01 12:01:57|info a|2015-01-01 12:00:00|
| y|2015-01-01 12:01:57|info b|2015-01-01 12:00:00|
+--------+-------------------+------+-------------------+
如果没有确定区间，那么只有交叉连接，然后找到min(timestampA - timestampB)区间就可以了。您可以使用窗口函数和row_number函数来实现这一点，如下所示：
w = Window.partitionBy('variable', 'info').orderBy(F.abs(F.col('timestampA').cast('int') - F.col('timestampB').cast('int')))
res = res.withColumn('rn', F.row_number().over(w)).filter('rn = 1').drop('rn')