从pySp中的dict构建行

Question

我试图在pySpark 1.6.1中动态构建一行，然后将其构建到数据帧中。一般的想法是将describe的结果扩展到包括倾斜和峰度。以下是我认为应该有效的：

from pyspark.sql import Row

row_dict = {'C0': -1.1990072635132698,
            'C3': 0.12605772684660232,
            'C4': 0.5760856026559944,
            'C5': 0.1951877800894315,
            'C6': 24.72378589441825,
            'summary': 'kurtosis'}

new_row = Row(row_dict)

但这会返回TypeError: sequence item 0: expected string, dict found，这是一个相当明显的错误。然后我发现如果我首先定义行字段，我可以使用dict：

r = Row('summary', 'C0', 'C3', 'C4', 'C5', 'C6')
r(row_dict)
> Row(summary={'summary': 'kurtosis', 'C3': 0.12605772684660232, 'C0': -1.1990072635132698, 'C6': 24.72378589441825, 'C5': 0.1951877800894315, 'C4': 0.5760856026559944})

这将是一个很好的步骤，只是看起来我不能动态地指定Row中的字段。我需要这个来处理未知行数和未知名称。根据文档，您实际上可以走另一条路：

>>> Row(name="Alice", age=11).asDict() == {'name': 'Alice', 'age': 11}
True

看来我应该能做到。可能还有一些旧版本中不推荐的特性允许这样做，例如here。有没有比我现在失踪的同龄人更多的？

Answer 1

如果dict不平坦，可以递归地将dict转换为行。

def as_row(obj):
    if isinstance(obj, dict):
        dictionary = {k: as_row(v) for k, v in obj.items()}
        return Row(**dictionary)
    elif isinstance(obj, list):
        return [as_row(v) for v in obj]
    else:
        return obj

Answer 2

可以按如下方式使用关键字参数解包：

Row(**row_dict)

## Row(C0=-1.1990072635132698, C3=0.12605772684660232, C4=0.5760856026559944, 
##     C5=0.1951877800894315, C6=24.72378589441825, summary='kurtosis')

需要注意的是，地址problems with older Python versions是internally sorts data by key。

这种行为可能会在即将发布的版本中被删除-请参见SPARK-29748在PySpark SQL行创建中删除字段排序。一旦删除，您就必须确保dict中的值顺序在所有记录中是一致的。