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.