Pyspark 数据框到 python 列表答案

【问题标题】：Pyspark dataframe to python listPyspark 数据框到 python 列表
【发布时间】：2021-11-14 23:24:44
【问题描述】：

我有以下 pyspark 数据框。

Job_name start_time status
A        09:00:00   Not started
B        09:30:00   Completed
C        09:30:00   Running

我正在尝试从上面的数据框创建一个列表，如下所示。

预期输出：

lst = ["job A Not started at 09:00:00", "job B Completed at 9:30", "job C Running at 9.30"]

有没有办法将 pyspark 数据帧转换为上面的列表？

【问题讨论】：

this topic 对您有帮助吗？
类似，但不完全相同。我需要一行的所有列都填充到单个输出列表中
你可以用现有列的连接构建一个额外的列，然后使用上面的方法

标签： python pyspark

【解决方案1】：

使用您的DF，稍微重新排列：

df = df.select('Job_name','status','start_time')

>>> df.show()
+--------+-----------+----------+
|Job_name|     status|start_time|
+--------+-----------+----------+
|       A|Not started|  09:00:00|
|       B|  Completed|  09:30:00|
|       C|    Running|  09:30:00|
+--------+-----------+----------+

尝试以下操作，使用list comprehension：

li = []
one = [li.append(' '.join(('Job',row[0],row[1],'at',row[2]))) for row in df.collect()]

>>> li

Out[123]: 

['Job A Not started at 09:00:00',
 'Job B Completed at 09:30:00',
 'Job C Running at 09:30:00']

【讨论】：

谢谢你。按预期工作

【解决方案2】：

你可以试试df.rdd.collect()：

print(['job {}{} at {}'.format(i,j,k) for i,j,k in zip([row['Job_name']],[row['status']], [row['start_time']]) for row in df.rdd.collect()])

应该做你正在寻找的东西。

【讨论】：

【解决方案3】：

您应该事先使用 spark 将所需的文件组合在一起，然后最后提取结果

数据准备

input_str = """
A|09:00:00   |Not started|
B|09:30:00   |Completed|
C|09:30:00   |Running
""".split("|")

input_values = list(map(lambda x: x.strip() if x.strip() != 'null' else None, input_str))

cols = list(map(lambda x: x.strip() if x.strip() != 'null' else None, "Job_name start_time status".split()))
            
n = len(input_values)
n_cols = 3

input_list = [tuple(input_values[i:i+n_cols]) for i in range(0,n,n_cols)]

sparkDF = sql.createDataFrame(input_list, cols)

sparkDF.show()

+--------+----------+-----------+
|Job_name|start_time|     status|
+--------+----------+-----------+
|       A|  09:00:00|Not started|
|       B|  09:30:00|  Completed|
|       C|  09:30:00|    Running|
+--------+----------+-----------+

收集

[x['output'] for x in 
    sparkDF.select(F.concat(
                    F.lit("job "),F.col('Job_name'),F.lit(" "),F.col("status"),F.lit(" at "),F.col('start_time')
                    ).alias("output")
              ).collect()

]

['job A Not started at 09:00:00',
 'job B Completed at 09:30:00',
 'job C Running at 09:30:00']

【讨论】：

不幸的是，我认为这不是 OP 所期望的输出。

【解决方案4】：

可以使用concat和collect_list函数实现

from pyspark.sql import functions as f

df = df.withColumn('ncol',f.concat_ws(" ",f.lit('Job'),f.col('Job_name'),f.col('status'),f.lit("at"),f.col('start_time')))
df.show(truncate=False)

+--------+----------+-----------+-----------------------------+
|Job_name|start_time|status     |ncol                         |
+--------+----------+-----------+-----------------------------+
|A       |09:00:00  |Not started|Job A Not started at 09:00:00|
|B       |09:30:00  |Completed  |Job B Completed at 09:30:00  |
|C       |09:30:00  |Running    |Job C Running at 09:30:00    |
+--------+----------+-----------+-----------------------------+

df.select(f.collect_list('ncol')).head()[0]

['Job B Completed at 09:30:00','Job C Running at 09:30:00','Job A Not started at 09:00:00']

【讨论】：