如何制作Pyspark行对象的突变副本

Question

from pyspark.sql import Row

一个行对象是不变的。它可以转换为Python字典，然后突变，然后返回行对象。有没有办法在不转换为字典的情况下进行可变或突变的副本，然后返回行？

这是在映射中运行的函数中的需要。

Answer 1

row.asDict()和 **dict都不保留字段的顺序。请注意，在Python 3.6 中，这可能会改变。请参阅PEP 468

类似于@hahmed所说的话。这动态创建了一个突变的行，但具有与行传递相同的模式。

from pyspark.sql import Row
from collections import OrderedDict
def copy(row, **kwargs):
    d = OrderedDict(zip(row.__fields__, row)) #note this is not recursive
    for key, value in kwargs.iteritems():
        d[key]=value
    MyRow = Row(row.__fields__)
    return MyRow(*d.values())

如果您需要将数据帧转换为RDD，然后将其再次成为DF

，这将很有用

例如

df_schema = df.schema
rdd = df_schema.rdd.map(lambda row: copy(row, field=newvalue))
new_df = spark.createDataFrame(rdd, df_schema)

Answer 2

这是制作我想出的突变副本的动态解决方案：

from pyspark.sql import Row
def copy(row, **kwargs):
    dict = {}
    for attr in list(row.__fields__):
        dict[attr] = row[attr]
    for key, value in kwargs.items():
        dict[key] = value 
    return Row(**dict)
row =  Row(name="foo", age=45)
print(row) #Row(age=45, name='foo')
new_row = copy(row, name="bar")
print(new_row) #Row(age=45, name='bar')

Answer 3

取决于您的实际用例，一种可能性是从现有的对象创建一个新行对象。

from pyspark.sql import Row
R = Row('a', 'b', 'c')
r = R(1,2,3)

假设我们要将a更改为r的3，从r中制作一个新的对象：

R(3, r.b, r.c)
# Row(a=3, b=2, c=3)

r仍然是：

r
# Row(a=1, b=2, c=3)