Извлечение значения из другой СТРОКИ в искровом фрейме данных

#scala #dataframe #apache-spark #apache-spark-sql

#scala #фрейм данных #apache-spark #apache-spark-sql

Вопрос:

У меня есть приведенный ниже DF со мной

  ------ ------ ---- 
|  Year|    PY| VAL|
 ------ ------ ---- 
|202005|201905|2005|
|202006|201906|2006|
|202007|201907|2007|
|201905|201805|1905|
|201906|201806|1906|
|201907|201807|1907|
|201805|201705|1805|
|201806|201706|1806|
|201807|201707|1807|
 ------ ------ ----

получено с помощью

 val df1=Seq(
("202005","201905","2005"),
("202006","201906","2006"),
("202007","201907","2007"),
("201905","201805","1905"),
("201906","201806","1906"),
("201907","201807","1907"),
("201805","201705","1805"),
("201806","201706","1806"),
("201807","201707","1807")
)toDF("Year","PY","VAL")

Я хотел бы заполнить значение предыдущего года (VAL_PY) в отдельном столбце. Это значение фактически находится в другой строке в том же DF.

Кроме того, я хотел бы добиться этого распределенным способом, поскольку мой DF большой (> 10 миллионов записей)

Ожидаемый результат —

  ------ ------ ---- ------- 
|  Year|    PY| VAL| VAL_PY|
 ------ ------ ---- ------- 
|202005|201905|2005|1905   |
|202006|201906|2006|1906   |
|202007|201907|2007|1907   |
|201905|201805|1905|1805   |
|201906|201806|1906|1806   |
|201907|201807|1907|1807   |
|201805|201705|1805|null   |
|201806|201706|1806|null   |
|201807|201707|1807|null   |
 ------ ------ ---- -------

1. Какова логика заполнения новых значений столбцов?

2. VAL_PY — это значение предыдущего года, которое находится в том же DF, но в другой строке

3. Да, но как вы определяете эту строку?

4. Например, в первой СТРОКЕ у нас есть year = 202005 и PY = 201905. Следовательно, VAL_PY = VAL, когда год = 201905

Ответ №1:

 val df1=Seq(("202005","201905","2005"),("202006","201906","2006"),("202007","201907","2007"),("201905","201805","1905"),("201906","201806","1906"),("201907","201807","1907"),("201805","201705","1805"),("201806","201706","1806"),("201807","201707","1807")
)toDF("Year","PY","VAL")

val df2 = df1
.drop("PY")
.withColumnRenamed("VAL","VAL_PY")
.withColumnRenamed("Year","PY")

df1.join(df2, Seq("PY"),"left")
.select("Year","PY","VAL","VAL_PY").show

ВЫВОД :

  ------ ------ ---- ------ 
|  Year|    PY| VAL|VAL_PY|
 ------ ------ ---- ------ 
|202005|201905|2005|  1905|
|202006|201906|2006|  1906|
|202007|201907|2007|  1907|
|201905|201805|1905|  1805|
|201906|201806|1906|  1806|
|201907|201807|1907|  1807|
|201805|201705|1805|  null|
|201806|201706|1806|  null|
|201807|201707|1807|  null|
 ------ ------ ---- ------

Похоже на левое самосоединение. Пожалуйста, дайте мне знать, если я что-то упустил.

1. Но знаете ли вы, почему это не удается — val df2 = df1.drop(«PY»); df1.join(df2,df1(«PY») === df2(«VAL»))

2. Мне это кажется действительно интересным. : o

3. Какую ошибку вы получаете? С точки зрения синтаксиса это правильно. Логически это не будет соответствовать as PY и VAL , поскольку нет общих значений, следовательно, вы должны получить пустой DF

4. извините!! Я имел в виду это . Ранее была опечатка . значение df2 = df1.drop(«PY»); df1.join(df2,df1(«PY») === df2(«Год»))

5. Ошибка — используйте синтаксис ПЕРЕКРЕСТНОГО СОЕДИНЕНИЯ, чтобы разрешить декартовы произведения между этими отношениями.; Но когда я использовал — spark.conf.set(«spark.sql.crossJoin.enabled», «true»), я получаю NULL в правом DF .