Pyspark 在数据框中合并 WrappedArrays

Posted 2023-04-15

【中文标题】Pyspark 在数据框中合并 WrappedArrays

【英文标题】：Pyspark Merge WrappedArrays Within a Dataframe

【发布时间】：2017-09-18 22:31:20

【问题描述】：

当前 Pyspark 数据帧具有以下结构（col2 的 WrappedArrays 列表）：

+---+---------------------------------------------------------------------+  
|id |col2                                                                 |   
+---+---------------------------------------------------------------------+  
|a  |[WrappedArray(code2), WrappedArray(code1, code3)]                    |  
+---+---------------------------------------------------------------------+  
|b  |[WrappedArray(code5), WrappedArray(code6, code8)]                    |  
+---+---------------------------------------------------------------------+  

这是我想要的结构（col2 的扁平列表）：

+---+---------------------------------------------------------------------+  
|id |col2                                                                 |   
+---+---------------------------------------------------------------------+  
|a  |[code2,code1, code3)]                                                |  
+---+---------------------------------------------------------------------+  
|b  |[code5,code6, code8]                                                 |  
+---+---------------------------------------------------------------------+  

但我不确定如何进行这种转换。我试图做一个平面图，但似乎没有用。有什么建议吗？

【问题讨论】：

flatMap 是错误的做法。平面地图复制行。试试地图

【参考方案1】：

您可以使用 udf 和 rdd 两种方式来执行此操作。这是示例：-

df = sqlContext.createDataFrame([
    ['a',  [['code2'],['code1', 'code3']]],  
    ['b',  [['code5','code6'], ['code8']]]
], ["id", "col2"])  
df.show(truncate = False)
+---+-------------------------------------------------+
|id |col2                                             |
+---+-------------------------------------------------+
|a  |[WrappedArray(code2), WrappedArray(code1, code3)]|
|b  |[WrappedArray(code5, code6), WrappedArray(code8)]|
+---+-------------------------------------------------+

RDD:-

df.map(lambda row:(row[0], reduce(lambda x,y:x+y, row[1]))).toDF().show(truncate=False)
+---+---------------------+
|_1 |_2                   |
+---+---------------------+
|a  |[code2, code1, code3]|
|b  |[code5, code6, code8]|
+---+---------------------+

UDF:-

from pyspark.sql import functions as F
import pyspark.sql.types as T
def fudf(val):
    #emlist = []
    #for item in val:
    #    emlist += item
    #return emlist
    return reduce (lambda x, y:x+y, val)
flattenUdf = F.udf(fudf, T.ArrayType(T.StringType()))
df.select("id", flattenUdf("col2").alias("col2")).show(truncate=False)
+---+---------------------+
|id |col2                 |
+---+---------------------+
|a  |[code2, code1, code3]|
|b  |[code5, code6, code8]|
+---+---------------------+

【讨论】：

谢谢！这些很棒，绝对可以解决问题。一个小的更正：请将 RDD 示例更新为 df.rdd.map() 而不是 df.map()。

@Eka 既然回答解决了您的问题，请接受（回答占用了受访者宝贵的时间）