Count Distinct与季度聚合

Question

我将一些日常数据存储在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')
)

然后，我连接并聚合包含产品列表的第二个dataframe。

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

上面的汇总让我可以看到自2018-11-01以来使用每个x_id的每个产品的h_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

预期输出将与我的代码生成的输出相同，但增加了一年/季度分组。

Answer 1

您还可以按季度进行分组，在日期上使用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')
    )
)