我在 pyspark 中有兩個數(shù)據(jù)框。如下所示，df1 包含來自傳感器的整個 long_lat。第二個數(shù)據(jù)幀 df2 是第一個數(shù)據(jù)幀的子集，其中 lat-long 值四舍五入為小數(shù)點后 2 位，然后刪除重復(fù)項以保留唯一的 lat_long 數(shù)據(jù)點。df1:+-----------------+---------+-----+--------------------+----------+------------+| UID| label|value| datetime| latitude| longitude|+-----------------+---------+-----+--------------------+----------+------------+|1B0545GD6546Y|evnt | 3644|2020-06-08T23:32:...|40.1172005|-105.0823546||1B0545GD6FG67|evnt | 3644|2020-06-08T23:32:...|40.1172201|-105.0821007||15GD6546YFG67|evnt | 3644|2020-06-08T23:32:...|40.1172396|-105.0818468||1BGD6546YFG67|evnt | 3644|2020-06-08T23:32:...|40.1172613|-105.0815929||1BGD6546YFG67|evnt | 3644|2020-06-08T23:32:...|40.1172808|-105.0813368||1B054546YFG67|evnt | 3644|2020-06-08T23:32:...|40.1173003|-105.0810742||1B056546YFG67|evnt | 3644|2020-06-08T23:32:...| 40.117322|-105.0808073|df2:+-------+--------+----------------+--------------+ |new_lat|new_long| lat_long| State_name|+-------+--------+----------------+--------------+| 40.13| -105.1|[40.13, -105.1] | Colorado|| 40.15| -105.11|[40.15, -105.11]| Colorado|| 40.12| -105.07|[40.12, -105.07]| Colorado|| 40.13| -104.99|[40.13, -104.99]| Colorado|| 40.15| -105.09|[40.15, -105.09]| Colorado|| 40.15| -105.13|[40.15, -105.13]| Colorado|| 40.12| -104.94|[40.12, -104.94]| Colorado|因此，df2 的行數(shù)比第一個少得多。在 df2 中，我應(yīng)用了一個 udf 來計算州名稱。現(xiàn)在我想在 df1 中填充州名稱。由于 df2 的 lat_long 值四舍五入為小數(shù)點后 2 位，為了匹配我使用如下閾值，我在這里使用連接操作。threshold = 0.01df4 = df1.join(df2)\ .filter(df2.new_lat - threshold < df1.latitude)\ .filter(df1.latitude < df2.new_lat + threshold)有沒有其他有效的方法來實現(xiàn)同樣的目標(biāo)？因為連接操作是做笛卡爾積，它需要時間和大量的任務(wù)。考慮一下，我的 df1 將有 10000 億條記錄。任何幫助將不勝感激。
1 回答
慕標(biāo)琳琳
TA貢獻1830條經(jīng)驗 獲得超9個贊
每當(dāng)你連接一個大的 DataFrame 和一個小的 DataFrame 時，你應(yīng)該總是嘗試執(zhí)行廣播連接。
如果df2小到可以廣播，那么df1.join(broadcast(df2))性能會更好。
該方法的第二個參數(shù)join()應(yīng)該是連接條件。
def approx_equal(col1, col2, threshold):
return abs(col1 - col2) < threshold
threshold = lit(0.01)
df4 = df1.join(broadcast(df2), approx_equal(df2.new_lat, df1.latitude, threshold) && approx_equal(df2.new_long, df1. longitude, threshold))
編輯：我將approx_equal函數(shù)添加到quinn，所以你的代碼可以更簡潔：
import quinn as Q
threshold = lit(0.01)
df4 = df1.join(broadcast(df2), Q.approx_equal(df2.new_lat, df1.latitude, threshold) && Q.approx_equal(df2.new_long, df1. longitude, threshold))
添加回答
舉報
0/150
提交
取消