Как добавить строки одного столбца кадра данных и сформировать другой столбец, который будет иметь инкрементное значение исходного столбца

Question

У меня есть DataFrame, данные которого я вставляю ниже:

+---------------+--------------+----------+------------+----------+
|name           |      DateTime|       Seq|sessionCount|row_number|
+---------------+--------------+----------+------------+----------+
|            abc| 1521572913344|        17|           5|         1|
|            xyz| 1521572916109|        17|           5|         2|
|           rafa| 1521572916118|        17|           5|         3|
|             {}| 1521572916129|        17|           5|         4|
|     experience| 1521572917816|        17|           5|         5|
+---------------+--------------+----------+------------+----------+

Столбец 'name' имеет тип string. Я хочу новый столбец "effective_name", который будет содержать дополнительные значения "name", как показано ниже:

+---------------+--------------+----------+------------+----------+-------------------------+
|name          | DateTime |sessionSeq|sessionCount|row_number |effective_name|
+---------------+--------------+----------+------------+----------+-------------------------+
|abc            |1521572913344 |17        |5           |1         |abc                      |
|xyz            |1521572916109 |17        |5           |2         |abcxyz                   |
|rafa           |1521572916118 |17        |5           |3         |abcxyzrafa               |
|{}             |1521572916129 |17        |5           |4         |abcxyzrafa{}             |
|experience     |1521572917816 |17        |5           |5         |abcxyzrafa{}experience   |
+---------------+--------------+----------+------------+----------+-------------------------+

Новый столбец содержит добавочную конкатенацию своих предыдущих значений столбца name.

Answer 1

Решение:

import pyspark.sql.functions as f

w = Window.partitionBy ("Seq"). OrderBy ("DateTime")

df.select ( "*", f.concat_ws ( "", f.collect_set (f.col ( "имя")). над (ш) ) .alias ( "cummuliative_name") ) .Show ()

Объяснение

collect_set () - эта функция возвращает значение наподобие [["abc", "xyz", "rafa", {}, "experience"]].

concat_ws () - эта функция принимает выходные данные collect_set () в качестве входных данных и преобразует их в abc, xyz, rafa, {}, experience

Примечание: Используйте collect_set () , если у вас нет дубликатов, или используйте collect_list ()

Answer 2

Вы можете достичь этого, используя pyspark.sql.Window, который упорядочивает по clientDateTime, pyspark.sql.functions.concat_ws и pyspark.sql.functions.collect_list:

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

w = Window.orderBy("DateTime")  # define Window for ordering

df.drop("Seq", "sessionCount", "row_number").select(
    "*",
    f.concat_ws(
        "",
        f.collect_list(f.col("name")).over(w)
    ).alias("effective_name")
).show(truncate=False)
#+---------------+--------------+-------------------------+
#|name           |      DateTime|effective_name           |
#+---------------+--------------+-------------------------+
#|abc            |1521572913344 |abc                      |
#|xyz            |1521572916109 |abcxyz                   |
#|rafa           |1521572916118 |abcxyzrafa               |
#|{}             |1521572916129 |abcxyzrafa{}             |
#|experience     |1521572917816 |abcxyzrafa{}experience   |
#+---------------+--------------+-------------------------+

Я опустил "Seq", "sessionCount", "row_number", чтобы сделать вывод на экран более дружественным.

Если вам нужно сделать это для каждой группы, вы можете добавить partitionBy к Window. Скажем, в этом случае вы хотите сгруппировать по sessionSeq, вы можете сделать следующее:

w = Window.partitionBy("Seq").orderBy("DateTime")

df.drop("sessionCount", "row_number").select(
    "*",
    f.concat_ws(
        "",
        f.collect_list(f.col("name")).over(w)
    ).alias("effective_name")
).show(truncate=False)
#+---------------+--------------+----------+-------------------------+
#|name           |      DateTime|sessionSeq|effective_name           |
#+---------------+--------------+----------+-------------------------+
#|abc            |1521572913344 |17        |abc                      |
#|xyz            |1521572916109 |17        |abcxyz                   |
#|rafa           |1521572916118 |17        |abcxyzrafa               |
#|{}             |1521572916129 |17        |abcxyzrafa{}             |
#|experience     |1521572917816 |17        |abcxyzrafa{}experience   |
#+---------------+--------------+----------+-------------------------+

Если вы предпочитаете использовать withColumn, вышеуказанное эквивалентно:

df.drop("sessionCount", "row_number").withColumn(
    "effective_name",
    f.concat_ws(
        "",
        f.collect_list(f.col("name")).over(w)
    )
).show(truncate=False)

---

Объяснение

Вы хотите применить функцию к нескольким строкам, которая называется агрегацией. При любой агрегации вам нужно определить, какие строки агрегировать и в каком порядке. Мы делаем это, используя Window. В этом случае w = Window.partitionBy("Seq").orderBy("DateTime") разделит данные по Seq и отсортирует по DateTime.

Сначала мы применяем агрегатную функцию collect_list("name") к окну. Это собирает все значения из столбца name и помещает их в список. Порядок вставки определяется порядком окна.

Например, промежуточный результат этого шага будет:

df.select(
    f.collect_list("name").over(w).alias("collected")
).show()
#+--------------------------------+
#|collected                       |
#+--------------------------------+
#|[abc]                           |
#|[abc, xyz]                      |
#|[abc, xyz, rafa]                |
#|[abc, xyz, rafa, {}]            |
#|[abc, xyz, rafa, {}, experience]|
#+--------------------------------+

Теперь, когда соответствующие значения находятся в списке, мы можем объединить их вместе с пустой строкой в ​​качестве разделителя.

df.select(
    f.concat_ws(
        "",
        f.collect_list("name").over(w)
    ).alias("concatenated")
).show()
#+-----------------------+
#|concatenated           |
#+-----------------------+
#|abc                    |
#|abcxyz                 |
#|abcxyzrafa             |
#|abcxyzrafa{}           |
#|abcxyzrafa{}experience |
#+-----------------------+