How do I add an empty column with a certain schema to a PySpark dataframe?

Viewed 122

The problem:

I want to transform some data that is being inconsistently created then sent to me. As an example, see the following table:

Id Column 1 Column 2
1 somedata somedata
2 somedata somedata
3 somedata somedata

Sometimes, when a certain column has no data for a given period of time (the refresh time I have to use), the data arrives with a missing column, instead of having an empty one. For example:

Expected:

Id Column 1 Column 2
1 somedata
2 somedata
3 somedata

What is happening:

Id Column 2
1 somedata
2 somedata
3 somedata

So I created the following function:

def fix_null_fields_in_df(df : DataFrame) -> DataFrame: 

df_fields = df.schema.fields

for field in df_fields:
    if(field.dataType.typeName() == "struct"):
        print("is struct")

        exploded_df = df.select(field.name + ".*")
        print("exploded " + field.name)
        exploded_df = fix_null_fields_in_df(exploded_df)
        print("fixed " + field.name)
        print(exploded_df.schema)

        df = df.drop(field.name)
        exploded_df = spark.createDataFrame(data=[],schema=exploded_df.schema)
        #I could try creating a new dataframe and then joining it to the original one

        df = df.withColumn(
            field.name,
            struct(
               exploded_df.select("*")
            )
        )#this doesn't work, because I do not know how to cast without explicitly declaring column names

    else: 
        print("is not struct")

        if str(field.dataType) == 'NullType':
            print(field.name + " was null")
            df = df \
            .withColumn((field.name, df[str(field.name)].cast('string')))

return df

My idea is to recursively check for structs (since the data I'm working with has about 200 struct columns, some of which are nested), and explode every struct to check for null types (because that was causing some problems down the line). I hit a wall when trying to add a column with a defined schema to my original dataframe. Is it possible to add an arbitrary column to a dataframe?

All this is just to create a "template" dataframe with the desired schema (and no data), in order to add the missing columns.

0 Answers
Related