PySpark：如何在宽度可变的列数组上聚合？

Posted 2023-02-03

因此，我试图汇总并创建一系列手段（这是一个最小的工作示例：]

n = len(allele_freq_total.select("alleleFrequencies").first()[0])

allele_freq_by_site = allele_freq_total.groupBy("contigName", "start", "end", "referenceAllele").agg(
  array(*[mean(col("alleleFrequencies")[i]) for i in range(n)]).alias("mean_alleleFrequencies")

使用我从]获得的解决方案>

Aggregate over column arrays in DataFrame in PySpark?

但是问题是n是可变的，我该如何更改

array(*[mean(col("alleleFrequencies")[i]) for i in range(n)])

以便将可变长度考虑在内？

<<

答案

在不同组中具有不等大小的数组（对于您来说，一个组是("contigName", "start", "end", "referenceAllele")，我将其简单地重命名为group），您可以考虑展开数组列（alleleFrequencies），并进行介绍值在数组中的位置。这将为您提供一个额外的列，您可以在分组中使用该列来计算您想到的平均值。此时，您实际上可能已经有足够的空间来进行进一步的计算（请参见下面的df3.show()）。

如果真的必须将其放回数组中，那会比较困难，我也不知道。必须跟踪订单的顺序，我相信使用地图（如果需要，可以使用字典）很容易。为此，我在两列上使用了聚合函数collect_list。尽管collect_list不是确定性的（您不知道列表中返回值的顺序，因为行被乱序），但两个数组的聚合将保留其顺序，因为行将被整体重排（请参见下面的df4.show()）。在此处，您可以使用map_from_arrays创建位置到平均值的映射。