Python PySpark 中日期范围的递归计算和转置结果

这是另一个Spark-SQL解决方案.

假设op希望将给定Year_MONTER_NUMBER的12个月快照作为输入.

创建一个参考视图，其中包含从2010年开始到2025年的所有年月组合的空额值，即12*15.

val df = spark.read.format("csv").option("header","true").option("inferSchema","true").load("sale.csv")
val df2=df.withColumn("dt2",to_date(col("sale_date"),"MM/dd/yy"))
df2.createOrReplaceTempView("sale")

val ref_df = spark.sql(" select add_months('2010-01-01',id) yyyymm, cast(null as decimal(15,3)) amt from range(12*15)  order by 1 ")
ref_df.createOrReplaceTempView("ref")

spark.conf.set("spark.sql.crossJoin.enabled","true") // to allow cross join 
// Inputs
val year_month=202304
val month=year_month.toString.drop(4)
val year=year_month.toString.take(4)
val yearp=year.toInt-1
val dfs = spark.sql(s""" with t1 (select dept_id, date_format(dt2,'yyyyMM') yyyymm,  sum(sale_amt) sale_amt 
                                            from sale where year(dt2) in (${year},${yearp}) group by 1,2
                                   union all 
                                    select dept_id, date_format(yyyymm,'yyyyMM') yyyymm, amt from ref , (select distinct(dept_id) dept_id from sale) where year(yyyymm) in (${year},${yearp})
                                 )
 select dept_id, yyyymm, sale_amt amt from t1 
 where months_between(to_date('${year_month}01','yyyyMMdd'),to_date(yyyymm||'01','yyyyMMdd')) < 12
 and   months_between(to_date('${year_month}01','yyyyMMdd'),to_date(yyyymm||'01','yyyyMMdd')) >= 0   
""")

dfs.show(50,false)

dfs.createOrReplaceTempView("sale2")

val vl_ym = spark.sql("select collect_set(yyyymm) from sale2").as[Seq[String]].first.map( x => "'"+x+"'").sorted.reverse.mkString(",")
val dfs2 = spark.sql(s"""
select * from (select dept_id, yyyymm, sum(amt) amt from sale2 group by 1,2 ) t
 PIVOT ( 
  sum(amt) as amt
  FOR yyyymm IN ( ${vl_ym} )
  ) 
""")

dfs2.withColumn("year",lit(year)).withColumn("month",lit(month)).orderBy("dept_id").show(false)

最后一个输出是

+-------+------+------+------+-------+------+------+------+------+------+------+------+------+----+-----+
|dept_id|202304|202303|202302|202301 |202212|202211|202210|202209|202208|202207|202206|202205|year|month|
+-------+------+------+------+-------+------+------+------+------+------+------+------+------+----+-----+
|1      |70.000|45.000|12.000|100.000|40.000|40.000|75.000|80.000|65.000|70.000|null  |90.000|2023|04   |
|2      |70.000|45.000|12.000|100.000|40.000|40.000|80.000|30.000|null  |null  |null  |null  |2023|04   |
|3      |null  |null  |null  |null   |null  |null  |null  |50.000|65.000|70.000|null  |90.000|2023|04   |
+-------+------+------+------+-------+------+------+------+------+------+------+------+------+----+-----+

您可以重命名202304,202303.销售_1，销售_2..等

用于连接的字符串的spark 源部分:

>>> seqString=spark.sql("select array_sort(collect_set(yyyymm)) from ( select date_format(add_months(date'2023-01-01',id),'yyyyMM') yyyymm from range(12)) ").collect()[0][0]
>>> sorted_seq = sorted(seqString, reverse=True)
>>> addPrefix=''.join(map(lambda x: "'" + x + "'",sorted_seq))
>>> addPrefix
"'202312''202311''202310''202309''202308''202307''202306''202305''202304''202303''202302''202301'"
>>>