您如何从 pyspark 中的枢轴 table add/include header 行和总计行？

Question

我正在通过 PySpark 将数据导出到 Excel。我有一个数据集

df_raw = spark.createDataFrame([("2015-10", 'U.S.', 500), \
                                ("2018-10", 'Germany', 580), \
                                ("2019-08", 'Japan', 230), \
                                ("2015-12", 'U.S.', 500), \
                                ("2015-11", 'Germany', 580), \
                                ("2015-12", 'Japan', 502), \
                                ("2018-10", 'U.S.', 520), \
                                ("2019-08", 'Canada', 200)]).toDF("ym", "country", "points")
+-------+-------+------+
|     ym|country|points|
+-------+-------+------+
|2015-10|   U.S.|   500|
|2018-10|Germany|   580|
|2019-08|  Japan|   230|
|2015-12|   U.S.|   500|
|2015-11|Germany|   580|
|2015-12|  Japan|   502|
|2018-10|   U.S.|   520|
|2019-08| Canada|   200|
+-------+-------+------+

我将其转换为枢轴 table

df_pivot = df_raw.groupBy('country').pivot("ym").sum('points')
+-------+-------+-------+-------+-------+-------+
|country|2015-10|2015-11|2015-12|2018-10|2019-08|
+-------+-------+-------+-------+-------+-------+
|Germany|   null|    580|   null|    580|   null|
|   U.S.|    500|   null|    500|    520|   null|
| Canada|   null|   null|   null|   null|    200|
|  Japan|   null|   null|    502|   null|    230|
+-------+-------+-------+-------+-------+-------+

我想通过 Openpyxl.[=18= 将带有 header 行和总计行的 table 导出到 Excel 电子表格中]

我可以使用 .collect() 遍历数据框并将记录附加到工作表，但它不包括 header，我还想添加一个总计行。

总计行示例：

+-------+-------+-------+-------+-------+-------+
|country|2015-10|2015-11|2015-12|2018-10|2019-08|
+-------+-------+-------+-------+-------+-------+
|Germany|   null|    580|   null|    580|   null|
|   U.S.|    500|   null|    500|    520|   null|
| Canada|   null|   null|   null|   null|    200|
|  Japan|   null|   null|    502|   null|    230|
+-------+-------+-------+-------+-------+-------+
|       |    500|    580|   1002|   1100|    430|
+-------+-------+-------+-------+-------+-------+

我该如何完成？

Answer 1

尝试查看 rollup 函数并在之后合并它，例如

df = df_raw.groupBy('country').pivot("ym").sum('points')
df2 = df.rollup('country').count()

或者，只需获取数据透视表的输出，动态 select 日期列（在正则表达式模式或其他内容上）并将它们与 sum() 聚合，并将别名返回到列名称中。

编辑：现在我明白你到底想要什么了。我仍然会使用 rollup 但结合一些重命名和联合，例如：

from functools import reduce

agg_cols = df_pivot.columns[1:]
rollup_df = df_pivot.rollup().sum()

renamed_df = reduce(
    lambda rollup_df, idx: rollup_df.withColumnRenamed(rollup_df.columns[idx], agg_cols[idx]), 
    range(len(rollup_df.columns)), rollup_df
)

renamed_df = renamed_df.withColumn('country', f.lit('Total'))

df_pivot.unionByName(
    renamed_df
).show()

输出：

+-------+-------+-------+-------+-------+-------+
|country|2015-10|2015-11|2015-12|2018-10|2019-08|
+-------+-------+-------+-------+-------+-------+
|Germany|   null|    580|   null|    580|   null|
|   U.S.|    500|   null|    500|    520|   null|
| Canada|   null|   null|   null|   null|    200|
|  Japan|   null|   null|    502|   null|    230|
|  Total|    500|    580|   1002|   1100|    430|
+-------+-------+-------+-------+-------+-------+

在 PySpark 2.4.3 上测试

您如何从 pyspark 中的枢轴 table add/include header 行和总计行？

How do you add/include header row and totals row from a pivot table in pyspark?

openpyxl

pyspark