python - 将字段附加到 pyspark 中的一行

Question

我有一个 DataFrame 并在其上运行一个 flatMap。在 map 函数中，我试图将一个新字段附加到给定行。

我该怎么做？

def mapper(row):
    value = 0 #some computation here
    row.append(newvalue = value) #??? something like that
    return row

data = sqlContext.jsonFile("data.json")
mapped = data.flatMap(mapper)
#do further mappings with the new field

Answer 1

在您的指导下，我创造了一些更灵活的东西，希望它能有所帮助:

from pyspark.sql import Row

def addRowColumn(row, **kwargs):
    rowData = row.asDict()
    for column in kwargs:
        rowData[column] = kwargs[column]
    return Row(**rowData)

要在单行上使用它，只需这样调用:

modifiedRow = addRowColumn(originalRow, test="Hello Column!")

要在整个数据集上运行，只需创建一个 udf 来更改每一行。