将特定功能应用于星火数据框架中结构化列的有效途径？

Question

我在星火数据框架中有数据，列col包含表单的结构化数据：

------ col -------   # Column whose elements are structures
field0  field1 …     # StructType with StructFields (variable names and count)
[1,2,3] [4,5]  [6]   # Each field is of type ArrayType
[1,2]   [3]    []
…

字段的数目和名称不固定的。

计算每一行中元素总数的最有效方法是什么?在上面的示例中，中，预期的结果数据框架是这样的：

num_elements
6
3
…

始终有用户定义函数的解决方案：

from pyspark.sql.types import IntegerType

def num_elements(all_arrays_in_row):
    return sum(map(len, all_arrays_in_row))
num_elements = pyspark.sql.functions.udf(num_elements, IntegerType())

data_frame.select(num_elements(data_frame.col)).show()  # Number of elements in each row

现在，我不知道这是否普遍有效，因为：

1. 函数num_elements()是用Python编写的。
2. 如果由于某种原因而没有将字段存储在一起，则map()在计算每个数组的长度之前强制对其进行提取。

更广泛地说，“纯粹的”火花方式会更有效，但它是逃避我。到目前为止，我尝试了以下几点，但这比上面的方法要麻烦得多，而且也不完整：

1. 使用field0 (烦琐)获取字段名[field.name for field in data_frame.select("col").schema.fields[0].dataType.fields]等。
2. 对于每个字段名，有效地计算其数组的大小： sizes_one_field = data_frame.select(pyspark.sql.functions.size( data_frame.col.getField(Field_name)

现在，我被困在这一点上，因为我不知道如何将1列数据框架sizes_one_field (每个字段名有一个)相加。另外，也许有一种方法可以直接将size()函数应用于星火中col列的每个字段(通过某种映射？)还是一种完全不同的方法来获取每一行中的元素总数？

Answer 1

您可以尝试以下几种方法：

from pyspark.sql import functions as f

result = df.select(sum((f.size(df[col_name]) for col_name in df.columns), f.lit(0)))

该解决方案使用pyspark.sql内置函数，并将以优化的方式执行。有关这些函数的更多信息，您可以查看其电火花文档。