Создайте новый столбец на основе RECORD_EFFECTIVE_FROM_DATE с помощью pyspark. ⇐ Python

[Гость]

Мне нужно использовать pyspark. Создайте новый столбец с именем Record_efficient_to_Date, переместив значения из Record_efficient_From_Date. Я пробовал функции опережения и задержки, но они не заполняют правильный набор результатов. Как вы можете видеть, выделенная часть неверна при вводе.
Вывод правильный, также любые значения, которые повторяют идентификатор, добавляются как 6f,7g,7h и т. д.
Любые предложения по приведенным ниже требованиям.
Ввод –
[img]https://i.stack.imgur .com/xZYS9.jpg[/img]
Вывод исключен —
[img]https://i.stack.imgur. com/ZF91p.jpg[/img]

Код: Выделить всё

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window
import pandas as pd

# Your data
data = {
"USER_ID": [59515, 59515, 59515, 59515, 59515, 59515, 59515, 59515, 59515, 59515, 59515],
"RECORD_EFFECTIVE_FROM_DATE": ["4/14/2022", "4/15/2022", "4/26/2022", "5/30/2022", "2/17/2023", "8/4/2023", "12/8/2023", "12/8/2023", "1/5/2024", "1/19/2024", "1/19/2024"],
"IDENTIFIER": ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K']
}

# Create a PySpark DataFrame
df = spark.createDataFrame(pd.DataFrame(data))
display(df)
# Define a window specification to partition by USER_ID and order by RECORD_EFFECTIVE_FROM_DATE
window_spec = Window.orderBy("IDENTIFIER")
# Assign the shifted values directly to the new column
df = df.withColumn("RECORD_EFFECTIVE_TO_DATE", F.lead("RECORD_EFFECTIVE_FROM_DATE").over(window_spec))
df = df.withColumn("keycolumn", (F.monotonically_increasing_id() + 1).cast("int"))

df = df.withColumn("keycolumn", F.when(df["RECORD_EFFECTIVE_TO_DATE"] == F.lag(df["RECORD_EFFECTIVE_TO_DATE"]).over(window_spec), F.concat(F.col("keycolumn"), F.col("IDENTIFIER"))).otherwise(F.col("keycolumn")))

# Show the result
display(df)

outcome after trying lead function. Tricky part is that on effective date next rows values comes twice, then I need to generate a new row with identifier like 6F.




Источник: https://stackoverflow.com/questions/781 ... ng-pyspark