所以我對 pyspark 很陌生，但我仍然無法正確創建自己的查詢。我嘗試用谷歌搜索我的問題，但我只是不明白其中大部分是如何作業的。我不確定我錯過了什么。
但無論如何我有以下兩個資料框，spark_p：
-------------------- ----- --------------------
|__record_timestamp__|cycle| profiles|
-------------------- ----- --------------------
| 1651737406300000000| 0|[0, 1, 1, 1, 3, 1...|
| 1651737406300000000| 16|[0, 0, 1, 0, 0, 0...|
| 1651737406300000000| 17|[1, 1, 1, 1, 0, 0...|
| 1651737406300000000| 18|[0, 0, 0, 0, 0, 1...|
| 1651737406300000000| 19|[1, 1, 1, 0, 0, 0...|
-------------------- ----- --------------------
和 spark_m：
------------- --------------------
| current|__record_timestamp__|
------------- --------------------
| 0.007181627| 1651730407500000000|
| 8.3004625E-4| 1651730464000000000|
| 0.41976404| 1651730507000000000|
|-0.0017322368| 1651732761000000000|
|-2.5260705E-4| 1651732822500000000|
| 2.3460487E-4| 1651732824500000000|
------------- --------------------
我需要向 spark_p 添加一列，其中包含該特定時間戳的當前值。
所以結果看起來像：
-------------------- ----- -------------------- ---------
|__record_timestamp__|cycle| profiles| current|
-------------------- ----- -------------------- ---------
| 1651737406300000000| 0|[0, 1, 1, 1, 3, 1...| 0.07|
| 1651737406300000000| 16|[0, 0, 1, 0, 0, 0...| 12|
| 1651737406300000000| 17|[1, 1, 1, 1, 0, 0...| 0.0|
| 1651737406300000000| 18|[0, 0, 0, 0, 0, 1...| 5.235654|
| 1651737406300000000| 19|[1, 1, 1, 0, 0, 0...| 125|
-------------------- ----- -------------------- ---------
現在時間戳不會完全匹配，但我只需要最接近的時間戳，或者使用之前記錄的當前值，或者都可以。我不知道怎么...
當我嘗試：
spark_p.join(spark_m, spark_p.__record_timestamp__ == spark_m.__record_timestamp__, "inner").show()
我只是得到：
-------------------- ----- -------- ----- --------------------
|__record_timestamp__|cycle|profiles|value|__record_timestamp__|
-------------------- ----- -------- ----- --------------------
-------------------- ----- -------- ----- --------------------
所以我猜他們都不完全匹配，但我怎么能抓住最接近的值呢？TIA
uj5u.com熱心網友回復：
此解決方案包含答案：
SPLIT_COUNT = 90
SPLIT_SIZE = 1024
spark_p = data.select("profiles", '__record_timestamp__')
spark_p = spark_p.withColumn("profiles", F.col("profiles").getField("elements") )
slices = [F.slice(F.col('profiles'), i * SPLIT_SIZE 1, SPLIT_SIZE) for i in range(SPLIT_COUNT)]
spark_p = spark_p.select(F.posexplode(F.array(*slices)), F.col('__record_timestamp__'))
spark_p = spark_p.withColumn("cycle", F.col("pos") )
spark_p = spark_p.withColumn("profiles", F.col("col") )
spark_p = spark_p.drop('pos').drop('col')
spark_m = magnetData.select("value", '__record_timestamp__', )
spark_p = spark_p.withColumn('value', F.lit(None))
spark_m = spark_m.withColumn('profiles', F.lit(None))
spark_m = spark_m.withColumn('cycle', F.lit(None))
final_df = spark_p.unionByName(spark_m)
w = Window.orderBy('__record_timestamp__').rowsBetween(Window.unboundedPreceding, -1)
final_df = final_df.withColumn('value', F.last('value', True).over(w)).filter(~F.isnull('profiles'))
您必須使用 unboundedPreceding 引數創建一個視窗。
轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/474672.html
標籤：Python 阿帕奇火花 加入 pyspark apache-spark-sql
上一篇：如何在Java中加入兩個具有相同索引的ArrayList
下一篇：加入r中兩個資料幀的重疊范圍