使用循环 Pyspark 连接多列

【问题标题】：Concat multiple columns with loop Pyspark使用循环 Pyspark 连接多列
【发布时间】：2018-06-26 03:09:26
【问题描述】：

我有 n 个字符串列数组。我想使用循环将这 n 列连接为一个。

我有这个功能来连接列：

def concat(type):
    def concat_(*args):
        return list(chain(*args))
    return udf(concat_, ArrayType(type))

concat_string_arrays = concat(StringType())

在下面的例子中，我有 4 列，我将像这样连接：

df_aux = df.select('ID_col',concat_string_arrays(col("patron_txt_1"),col("patron_txt_2"),col('patron_txt_3'),col('patron_txt_0')).alias('patron_txt')

但是，如果我有 200 列，如何通过循环动态使用此函数？

【问题讨论】：

你能告诉我们你的数据是什么样的吗？以及最终结果应该如何。

标签： python python-2.7 pyspark

【解决方案1】：

您可以使用 * 运算符将列列表传递给您的 concat UDF：

from itertools import chain
from pyspark.sql.functions import col, udf
from pyspark.sql.types import *

df = sqlContext.createDataFrame([("1", "2","3","4"), 
                                 ("5","6","7","8")], 
                                 ('ID_col', 'patron_txt_0','patron_txt_1','patron_txt_2'))  

def concat(type):
    def concat_(*args):
        return list(chain(*args))
    return udf(concat_, ArrayType(type))


concat_string_arrays = concat(StringType())

#Select the columns you want to concatenate 
cols = [c for c in df.columns if c.startswith("patron_txt")]

#Use the * operator to pass multiple columns to concat_string_arrays
df.select('ID_col',concat_string_arrays(*cols).alias('patron_txt')).show()

这会产生以下输出：

+------+----------+
|ID_col|patron_txt|
+------+----------+
|     1| [2, 3, 4]|
|     5| [6, 7, 8]|
+------+----------+

【讨论】：