Блог

У меня есть некоторые ежедневные данные, хранящиеся в Spark dataframe, которые я агрегирую, чтобы получить количество. Я делаю это так:

 start = '2018-11-01'
end = '2021-02-19'

t1 = (
    spark.table('schema.t1')
    .where(sf.col('yyyy_mm_dd').between(start, end))
    .select('yyyy_mm_dd', 'x_id', 'h_id', 'app', 'kpi')
)
 

Затем я объединяю и объединяю со вторым фреймом данных, который содержит список продуктов.

 aggregate = (
    t1
    .join(t2, on = ['app', 'kpi'], how = 'left')
    .groupby('x_id', 'product')
    .agg(
        sf.countDistinct('h_id').alias('count_ever')
    )
)
 

Приведенная выше агрегация позволяет мне видеть отчетливое количество h_id использованных каждого продукта с тех пор 2018-11-01 , за x_id .

Мне было интересно, как я могу изменить агрегацию, чтобы она все еще выполнялась countDistinct() , но между датами начала и окончания квартала, а не за все время.

Поэтому вместо подсчета ever ( 2018-11-01 -> 2021-02-19 ), как это делает мой код, я бы посчитал эти диапазоны:

 2018-11-01 -> 2018-12-31
2019-01-01 -> 2019-03-31
2019-04-01 -> 2019-06-30
2019-07-01 -> 2019-09-30
2019-10-01 -> 2019-12-31
2020-01-01 -> 2020-03-31
2020-04-01 -> 2020-06-30
2020-07-01 -> 2020-09-30
2020-10-01 -> 2020-12-31 
2021-01-01 -> 2021-02-19
 

Ожидаемый результат будет таким же, как и в моем коде, но с дополнительной группировкой по годам / кварталам.

Вы также можете сгруппировать по кварталу, используя trunc по дате:

 aggregate = (
    t1
    .join(t2, on = ['app', 'kpi'], how = 'left')
    .groupby('x_id', 'product', sf.trunc(sf.to_date('yyyy_mm_dd', 'yyyy_MM_dd'), 'quarter').alias('quarter'))
    .agg(
        sf.countDistinct('h_id').alias('count_ever')
    )
)