Python, Select from Nested Dataframe when Input File Changes Format

Question

df = spark.read.json(['/Users/.../input/json/thisistheinputfile.json'])
df.printSchema()

Results in something like below:

root
 |-- _metadata: struct (nullable = true)
 |    |-- bundled: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- bundledIds: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- failedInitializations: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- unbundled: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |-- anonymousId: string (nullable = true)
 |-- channel: string (nullable = true)
 |-- context: struct (nullable = true)
 |    |-- campaign: struct (nullable = true)
 |    |    |-- content: string (nullable = true)
 |    |    |-- medium: string (nullable = true)
 |    |    |-- name: string (nullable = true)
 |    |    |-- source: string (nullable = true)
 |    |    |-- term: string (nullable = true)
 |    |    |-- utm_campaign: string (nullable = true)
 |    |    |-- utm_medium: string (nullable = true)
 |    |    |-- utm_term: string (nullable = true)
 |    |-- ip: string (nullable = true)

However (SOME TIME LATER) in some cases the input file does not contain some of the content that was present above, for instance, maybe the campaign information is not available:

root
 |-- _metadata: struct (nullable = true)
 |    |-- bundled: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- bundledIds: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- failedInitializations: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |    |-- unbundled: array (nullable = true)
 |    |    |-- element: string (containsNull = true)
 |-- anonymousId: string (nullable = true)
 |-- channel: string (nullable = true)
 |-- context: struct (nullable = true)
 |    |-- ip: string (nullable = true)

I would like to automatically be able to select some of the columns but I don't want the script to crash when the content is not available. Note that the number of columns that are to be selected are a lot more then in the example below:

df_2 = df\
    .select(expr("context.campaign.source").alias("campaign_source"),
            expr("context.campaign.utm_campaign").alias("utm_campaign"),
            'anonymousId')

One case could be that the anonymousId, ip and context.campaign.source exists, but not the context.campaign.utm_campaign and all the possible combinations (can be a lot with many columns).

I tried listing the part I wanted to find and check if they existed and could thereafter use that list as an input to the dataframe selection. But I found this difficult since I have a nested dataframe:

lst = ['anonymousId', 
        'source', 
        'utm_campaign', 
        'ip']
col_exists = []
for col in lst: 
    if df_seglog_prod.schema.simpleString().find(col) > 0:
        col_exists.append(col)
    else: 
        print('Column', col, 'does not exist')

df_2 = df.select(col_exsists) #does ofc not work...

Any tips on how to work with this kind of nested dataframe?

Thank you in advance!!

user_17682024 · Answer

The following steps helped resolve my issue:

def flatten(schema, prefix=None):
fields = []
for field in schema.fields:
    name = prefix + '.' + field.name if prefix else field.name
    dtype = field.dataType
    if isinstance(dtype, ArrayType):
        dtype = dtype.elementType
    if isinstance(dtype, StructType):
        fields += flatten(dtype, prefix=name)
    else:
        fields.append(name)

return fields

def intersection(lst1, lst2):
    # Use of hybrid method
    temp = set(lst2)
    lst3 = [value for value in lst1 if value in temp]
    return lst3

fieldsPathName = flatten(df.schema)
df_prepSchema = df.select(fieldsPathName).toDF(*fieldsPathName)

lst1 = ['context.campaign.source', 
        'context.campaign.utm_campaign', 
        'timestamp', 
        'anonymousId']

lst2 = df.columns
cols = intersection(lst1, lst2)

append_str = '`'
cols = [append_str + col for col in cols]
cols = [col + append_str for col in cols]
df_2 = df_prepSchema.select(cols)

Python, Select from Nested Dataframe when Input File Changes Format

1 Answers