如何在pyspark中使用groupby的聚合表达式列表？

我有一个包含 2 列的数据框，如下所示：

+----+---+
|ptyp|sID|
+----+---+
|  CO|111|
|  CO|222|
|  CO|222|
|  CO|222|
|  CO|111|
|  CD|111|
|  CD|222|
|  CD|222|
|  CD|333|
|  CD|333|
|  CD|333|
|  AG|111|
|  AG|111|
|  AG|111|
|  AG|222|
+----+---+

给定一个输入n，对于每个pytp，我想添加显示顶部的列n sIDs（就它们出现的次数而言pytp）。我还想打印每个组sID在列中出现的次数sIDval。

例如，如果n = 2，我希望输出是这样的：

+----+-------+-----------+-------+-----------+
|ptyp|topSID1|topSID1_val|topSID2|topSID2_val|
+----+-------+-----------+-------+-----------+
|  AG|    111|          3|    222|          1|
|  CO|    222|          3|    111|          2|
|  CD|    333|          3|    222|          2|
+----+-------+-----------+-------+-----------+

我UDFs用来计算这个：

@F.udf
def mode(x, top_rank):
    c = Counter(x).most_common(top_rank)
    sz = len(c)
    return c[min(top_rank-1, sz-1)][0]

@F.udf
def modeval(x, top_rank):
    c = Counter(x).most_common(top_rank)
    sz = len(c)
    return c[min(top_rank-1, sz-1)][1]

我将每个新列所需的聚合表达式存储在列表中newcols：

newcols = []
n = 3

for r in range(1, num_ranks+1):
    newcols.append([mode(F.collect_list('sID'), F.lit(r)).alias('topSID' + str(r))])
    newcols.append([modeval(F.collect_list('sID'), F.lit(r)).alias('topSID' + str(r) +'_val')])

现在，如果我知道n=3，我可以这样做：

df.groupBy('ptyp').agg(*newcols[0], *newcols[1], *newcols[2], \
                       *newcols[3], *newcols[4], *newcols[5])

有没有一种方法可以将其概括为 的任何值n？我试过了

df.groupBy('ptyp').agg([*e for e in new_cols])

和

df.groupBy('ptyp').agg((*e for e in new_cols))

以及更多变体，但所有变体都会出错。

现在我已经求助于一次聚合一个并进行连接，但这非常昂贵。

有没有办法以我上面尝试过的方式做到这一点？