Блог

У меня есть инкрементная загрузка в CSV-файлах. Я прочитал csv в фрейме данных. Фрейм данных содержит один столбец, содержащий несколько строк. Я должен найти отдельные строки из этого столбца и присвоить ID (целое число) каждому значению, начиная с 0 после присоединения к одному другому фрейму данных.

В следующем запуске я должен назначить идентификатор после определения максимального значения в ID столбце и увеличения его для разных строк. Везде, где в ID столбце есть null, я должен увеличить его ( 1) от значения предыдущего запуска.

ПЕРВЫЙ ЗАПУСК

ВТОРОЙ ЗАПУСК

MAX(ID) = 4

Я пробовал это, но не смог заставить это работать..

 max = df.agg({"ID": "max"}).collect()[0][0]
df_incremented = df.withcolumn("ID", when(col("ID").isNull(),expr("max  = 1")))
 

Дайте мне знать, если есть простой способ добиться этого.

Поскольку вы сохраняете только отдельные значения, вы можете использовать row_number функцию поверх window :

 from pyspark.sql import Window
from pyspark.sql import functions as F

 df = spark.createDataFrame(
    [("a",), ("a",), ("b",), ("c",), ("d",), ("e",), ("e",)],
    ("string",)
)

w = Window.orderBy("string")

df1 = df.distinct().withColumn("ID", F.row_number().over(w) - 1)

df1.show()
# ------ --- 
#|string| ID|
# ------ --- 
#|     a|  0|
#|     b|  1|
#|     c|  2|
#|     d|  3|
#|     e|  4|
# ------ --- 
 

Теперь давайте добавим несколько строк в этот фрейм данных и будем использовать row_number вместе с coalesce , чтобы присваивать ID только для строки, где она равна нулю (не нужно получать максимальное значение):

 df2 = df1.union(spark.sql("select * from values ('f', null), ('h', null), ('i', null)"))

df3 = df2.withColumn("ID", F.coalesce("ID", F.row_number(w) - 1))

df3.show()
# ------ --- 
#|string| ID|
# ------ --- 
#|     a|  0|
#|     b|  1|
#|     c|  2|
#|     d|  3|
#|     e|  4|
#|     f|  5|
#|     h|  6|
#|     i|  7|
# ------ --- 
 

Если вы хотите сохранить дублирующиеся значения и присвоить им одинаковые ID значения , используйте dense_rank вместо row_number .