我有幾個 PySpark 資料幀，其中第二個是從第一個派生的，用于檢查例外情況。
import pandas as pd
from datetime import date
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df1 = pd.DataFrame({
"pk": [1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3],
"date": [
date(2022,4,13), date(2022,4,14), date(2022,4,15), date(2022,4,16),
date(2022,4,13), date(2022,4,14), date(2022,4,15), date(2022,4,16),
date(2022,4,13), date(2022,4,14), date(2022,4,15), date(2022,4,16),
],
"varA": [4, 5, 1, 6, 7, 4, 8, 11, 12, 10, 11, 13],
"varB": [5, 6, 4, 3, 12, 13, 1, 14, 9, 10, 11, 15],
"varC": [8, 32, 1, 11, 4, 3, 5, 6, 10, 14, 9, 11]
})
df1 = spark.createDataFrame(df1)
所以第一個資料幀是一個“寬”資料幀。第二個是“長”資料幀，源自第一個例外處理演算法的結果。在這里，每一行都是一個例外。
df2 = do_anomaly_processing(df1)
df2.show()
# ----- ----------- -------- -----
#|pk | date|variable|value|
# ----- ----------- -------- -----
#| 1| 2022-04-14| varA| 5|
#| 2| 2022-04-14| varA| 4|
#| 3| 2022-04-14| varA| 10|
#| 3| 2022-04-15| varC| 14|
# ----- ----------- -------- -----
生成此資料幀的代碼是：
df2 = pd.DataFrame({
"pk": [1,2,3,3],
"date": [date(2022,4,14), date(2022,4,14), date(2022,4,14), date(2022,4,15)],
"variable": ["varA", "varA", "varA", "varC"],
"value": [5,4,10,14]
})
df2 = spark.createDataFrame(df2)
我想創建一個新的資料框，其中列出了每個例外的周圍值，如下所示：
# ----- ----------- -------- ----- ------ ------
#|pk | date|variable|value|v[n-1]|v[n 1]|
# ----- ----------- -------- ----- ------ ------
#| 1| 2022-04-14| varA| 5| 4| 1|
#| 2| 2022-04-14| varA| 4| 7| 8|
#| 3| 2022-04-14| varA| 10| 12| 11|
#| 3| 2022-04-15| varC| 9| 14| 11|
# ----- ----------- -------- ----- ------ ------
實際上，我希望能夠在資料允許的情況下在新資料框中放置盡可能多的過去和/或未來值（例如(v[n-5], v[n-4], v[n-3], v[n-2], v[n-1])，等 - 但始終是一個序列）。我最初是用一個for回圈來做這個的：
for step in dt:
if step == 0:
varName = "v[n]"
shiftedDataframe = melt(
df1,
id_vars=["pk", "date"],
value_vars=["varA", "varB", "varC"],
var_name=varName
)
else:
if step < 0:
varName = f"v[n{step}]"
step = abs(step)
elif step > 0:
varName = f"v[n {step}]"
step = -step
shiftedDataframe = melt(
create_shifted_dataframe(
df1,
"pk",
"date",
["varA", "varB", "varC"],
shiftAmount=step
),
id_vars=["pk", "date"],
value_vars=["varA", "varB", "varC"],
var_name=varName
)
df2 = df2.join(df1, on=["pk", "date", "variable"], how="left")
注意：功能melt和create_shifted_dataframe做你認為他們做的事。df1實際上，我df1正在創建原始資料幀df2( 這是我能找到的最好的方法，但必須有更有效的方法。我正在考慮做一個支點，然后是一個合并，但我不知道如何用df2. 無論如何，我希望這是有道理的，并且對某人來說是一個有趣的問題。
uj5u.com熱心網友回復：
如果我理解邏輯，您可以利用leadand lag。
melt但首先我們需要df1看起來像這樣，
--- ---------- -------- --------------
| pk| date|variable|original_value|
--- ---------- -------- --------------
| 1|2022-04-13| varA| 4|
| 1|2022-04-13| varB| 5|
...
如果你有這個資料框，你可以用df2.
df = df1.join(df2, on=['pk', 'date', 'variable'], how='left')
然后，您可以使用lead(get value at offset of the current row) 和lag(get value at offset at current row before) 獲得 v[n-1], v[n 1], v[n-2], v [n 2]...
# I will get until n -2 in this example.
w = Window.partitionBy(['pk', 'variable']).orderBy('date')
df = (df.select('pk', 'date', 'variable', 'value',
*[F.when(F.col('value').isNotNull(), F.lag('original_value', x).over(w)).alias(f'v[n-{x}]') for x in range(1, 3)],
*[F.when(F.col('value').isNotNull(), F.lead('original_value', x).over(w)).alias(f'v[n {x}]') for x in range(1, 3)])
.filter(F.col('value').isNotNull()))
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/468335.html
標籤：Python 阿帕奇火花 pyspark apache-spark-sql