PySpark如何将for循环输出的空行追加到DataFrame中

可以看到我们有两个DataFrame currdf 和 hist_df，逻辑已在代码中定义，预期输出为：

从预期输出中可以看到，每次迭代都会得到黄色标记的DataFrame，该临时表始终只有单行或者为空，我们需要将其持续追加到最终DataFrame中，如果临时DataFrame（tmp）为空，则仅存储实际的pcode值，其余字段留空。

复现代码 #

import pyspark
import os
import warnings
warnings.filterwarnings('ignore')
import pyspark.sql.functions as sf
from pyspark.sql import *
import pyspark
from pyspark.sql import SparkSession
spark=SparkSession.builder.appName('Pycode2Pyspark_RProducer').getOrCreate()
spark
currdf = spark.createDataFrame([(66, "d","4"), 
                                (67, "a", "0"), 
                                (70, "b", "4"), 
                                (71, "d", "4"),
                                (45, "x", "3"),
                                (48, "y", "3")],
                                ("id", "pcode", "amt"))
currdf.show()

hist_df = spark.createDataFrame([(66, "d",4,1), 
                                (67, "a", 0,0), 
                                (70, "b", 4,1), 
                                (71, "a", 9,0),
                                (45, "c", 2,1),
                                (12, "d", 7,0),
                                (37, "b", 3,0),
                                (89, "c", 1,0),
                                (11, "e", 9,1),
                                (79, "f", 6,1)],
                                ("id", "pcode", "amt","status1"))
hist_df.show()

import functools
from pyspark.sql import DataFrame
dataCollect = currdf.collect()
output_dfs = []
for row in dataCollect:
    temp_var =  row['pcode']
    print(temp_var)    
    temp_filter =  hist_df.where((hist_df['pcode'] == temp_var)) # Filter 
    temp_filter.show()
    tmp = temp_filter.groupby('pcode').agg(sf.sum('amt').alias('amt'),sf.sum('status1').alias('status1'))
    tmp = tmp.withColumn('status', sf.when((sf.col('amt')> 3) & (sf.col('status1')> 0),'Yes').otherwise('No'))
    tmp.show()
    output_dfs.append(tmp)
df_output = functools.reduce(DataFrame.union, output_dfs)

内容的提问来源于stack exchange，提问作者Nabi Shaikh