Блог

У меня есть образец данных, в которых я хочу приписать недостающие значения. Строки, в которых отсутствуют данные, обозначаются blank символом . Вот пример данных —

 val my_df = spark.sql(s"""
select 1 as id, 1 as time_gmt, 'a' as pagename
union
select 1 as id, 2 as time_gmt, 'b' as pagename
union
select 1 as id, 3 as time_gmt, 'blank' as pagename
union
select 1 as id, 4 as time_gmt, 'blank' as pagename
union
select 1 as id, 5 as time_gmt, 'd' as pagename
union
select 2 as id, 1 as time_gmt, 'c' as pagename
union
select 2 as id, 2 as time_gmt, 'a' as pagename
union
select 2 as id, 3 as time_gmt, 'c' as pagename
union
select 2 as id, 4 as time_gmt, 'blank' as pagename
union
select 2 as id, 5 as time_gmt, 'd' as pagename
""")
my_df.createOrReplaceTempView("my_df")

scala> my_df.orderBy("id","time_gmt").show(false)
 --- -------- -------- 
|id |time_gmt|pagename|
 --- -------- -------- 
|1  |1       |a       |
|1  |2       |b       |
|1  |3       |blank   |
|1  |4       |blank   |
|1  |5       |d       |
|2  |1       |c       |
|2  |2       |a       |
|2  |3       |c       |
|2  |4       |blank   |
|2  |5       |d       |
 --- -------- -------- 
 

Как вы можете видеть, есть 2 пробела для данных с идентификатором 1 и 1 пробел для данных с идентификатором 2. Я хочу заполнить эти значения, используя последнее непустое значение, наблюдаемое для каждого идентификатора, упорядоченное по time_gmt столбцу. Итак, мой вывод будет —

  --- -------- -------- ---------------- 
|id |time_gmt|pagename|pagename_imputed|
 --- -------- -------- ---------------- 
|1  |1       |a       | a              | 
|1  |2       |b       | b              | 
|1  |3       |blank   | b              | 
|1  |4       |blank   | b              |
|1  |5       |d       | d              | 
|2  |1       |c       | c              | 
|2  |2       |a       | a              | 
|2  |3       |c       | c              | 
|2  |4       |blank   | c              | 
|2  |5       |d       | d              | 
 --- -------- -------- ---------------- 
 

Как я могу это сделать в spark SQL?

ПРИМЕЧАНИЕ — пробелы могут появляться несколько раз для каждого раздела после непустых значений.

Один из вариантов использует оконные функции. Идея состоит в том, чтобы определить группы записей, где «пустые» записи будут принадлежать той же группе, что и последняя непустая.

Предполагая, что под пустым вы подразумеваете null , мы можем определить группы с количеством окон:

 select id, time_gmt, 
    max(pagename) over(partition by id, grp) as pagename
from (
    select t.*, 
        count(pagename) over(partition by id order by time_gmt) as grp
    from mytable t
) t
 

Если вы действительно имеете в виду строку 'blank' , то:

 select id, time_gmt, 
    max(case when pagename <> 'blank' then pagename end) over(partition by id, grp) as pagename
from (
    select t.*, 
        sum(case when pagename = 'blank' then 0 else 1 end) over(partition by id order by time_gmt) as grp
    from mytable t
) t
 

last(expr, [IgnoreNulls]) over() Для таких случаев использования существует специальная оконная функция:

 select id, time_gmt, last(nullif(pagename, 'blank'), true) over(partition by id order by time_gmt) as pagename
from my_df
 

https://spark.apache.org/docs/latest/api/sql/index.html#last