使用条件返回PySpark DataFrame的行

Question

我有两个长度不相同的数据帧。第一个看起来像这样

A      | Time_Stop             | B
------ | ----------------------|-----
Green  | 2016-10-01 00:10:15   | 77
Yellow | 2016-10-03 00:11:15   | 80
Blue   | 2016-10-04 00:12:15   | 6 

第二个是这样的

D      | Time_Start            | Z
------ | ----------------------|-----
Foo    | 2016-10-01 00:12:15   | 7
Cookie | 2016-10-03 00:45:15   | 99

我的目标是只返回在一定时间限制内(比如在5分钟内)的第一个数据帧中的行，因此输出帧应该看起来像这样

A      | Time_Stop             | B
------ | ----------------------|-----
Green  | 2016-10-01 00:10:15   | 77

我搞不懂这个。到目前为止，我已经尝试过了

from pyspark.sql import functions as F
timeFmt = "yyyy-MM-dd' 'HH:mm:ss"
result = df.where(F.unix_timestamp(df1.Time_Start, format = timeFmt) - F.unix_timestamp(df.Time_Stop, format = timeFmt) <= 300)

然而，这不起作用。我怎样才能达到我想要的结果呢?

编辑:我忘了提到两个dataframe的时间列都是字符串格式。

编辑2:我已经尝试了以下，并收到错误。

from pyspark.sql.functions import expo
df2 = df2.withColumn("Time_Start", df2["Time_Start"].cast("timestamp"))
df = df.withColumn("Time_Stop", df['Time_Stop'].cast('timestamp'))
condition = df.Time_Stop + expr("INTERVAL 10 MINUTES") <= df2.Time_Start
df.filter(condition).show()
AnalysisException: u'resolved attribute(s) starttime#2251 missing from pickup_time#1964,dropoff_latitude#2090,tip#2180,dropoff_longitude#2072,pickup_latitude#2036,pickup_longitude#2018,payment_type#2108,dropoff_time#2268,mta_tax#2162,trip_distance#2000,fare_amount#2126,toll#2198,rate_code#2054,total#2216,row#1946,surcharge#2144 in operator !Filter (cast(dropoff_time#2268 + interval 10 minutes as timestamp) <= starttime#2251);'

编辑3:我能够通过使用我的本地机器工作，但是我不认为我的代码会很好地翻译，当我转移到集群上运行。这是我的代码，也许有人可以指出方法，使它运行得更快或只是看起来更干净。我仍然让这个问题悬而未决。

df = list(df.toLocalIterator())
df1 = list(df1.toLocalIterator())
rand = []
for i in df:
    for j in df1:
        elapsed_time = (i['Time_Start'] - j['Time_Stop']).total_seconds()
        time_limit = 600
        if (abs(elapsed_time) <= time_limit):
            rand.append(j)
rand = list(set(rand))

Answer 1

使用toLocalIterator()与list()(其工作原理与collect()完全相同)和循环将在大数据集上非常低效(它根本不使用火花容量)。

在这种情况下，

笛卡尔连接似乎是最好的解决方案。让我们将包含Time_Stop的DF称为firstDF，将包含Time_Start的DF称为secondDF，两者都将日期转换为时间戳。然后试试下面的命令:

from pyspark.sql import functions as F
interval = F.unix_timestamp(secondDF.Time_Start) - F.unix_timestamp(firstDF.Time_Stop)
firstDF.join(secondDF).where(F.abs(interval) < 300).select('A', 'Time_Stop', 'B')