How to append empty row(for loop output) to a data frame in pyspark

user_5782481 0
Viewed 511

Data set Image

as one can observe we have two data frame currdf and hist_df, logic is defined in code and expected output is

Output

So in the expected output , one can observe for every iteration we get yellow marked data frame , always there is single row or no row in it we need to keep on appending to final data frame , and if data frame(tmp) is empty then only pcode will be store as actual and rest columns would be empty.

Reproduce Code

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)
pyspark apache-spark-sql dataframe for-loop python
1 Answers
from pyspark.sql import DataFrame
dataCollect = claimdf.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()
    if tmp.rdd.isEmpty():
        def convert(list):
            return tuple(list)
        columns = tmp.columns
        mylist = ['' for i in range(0,len(columns))]        
        element_index = [i[0] for i in  enumerate(columns) if i[1] == 'pcode'][0]
        mylist[element_index] =  temp_var
        mylist = convert(mylist)
        newRow = sqlContext.createDataFrame([mylist], columns)
        #newRow.show()
        output_dfs.append(newRow)
    else:
        output_dfs.append(tmp)
df_output = functools.reduce(DataFrame.union, output_dfs)
user_57824810
Related