我有一些日常数据存储在一个spark数据框中，我将其聚合以获得一个计数。我是这样做的：

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进行连接和聚合，该Dataframe包含产品列表。

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() 但在季度开始和结束日期之间，而不是所有时间。
所以我不再数了( 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

预期输出将与代码生成的输出相同，但有一个额外的年度/季度分组。