Resolved: Dataframe – splitting dictionary/map column’s keys and assigning values to each of the keys in a row. Pyspark/Pandas

              
CustomerID              counts
3004000304    {'MAJOR APPLIANCES, Home Office' -> 2, 'HOME OFFICE, ACCESSORIES' -> 1, 'OTHER STUFF' -> 2}
3004002756    {'MAJOR APPLIANCES, ACCESSORIES' -> 2}
3004000304    {'ACCESSORIES, OTHER STUFF' -> 2}

CustomerID              counts
3004000304    {'MAJOR APPLIANCES' -> 2, 'Home Office' -> 2, 'HOME OFFICE' -> 1, 'ACCESSORIES' -> 3, 'OTHER STUFF':4}
3004002756    {'MAJOR APPLIANCES' -> 2, 'ACCESSORIES' -> 2}

CustomerID              counts
3004000304    {'MAJOR APPLIANCES, Home Office' : 2, 'HOME OFFICE, ACCESSORIES' : 1, 'OTHER STUFF' : 2}
3004002756    {'MAJOR APPLIANCES, ACCESSORIES' : 2}

f = lambda x: dict(zip(x['Description'], x['Counts']))
df = categories.groupby(level=0).apply(f).to_frame('Counts')
print (df)

df.groupBy("CustomerID").agg(
    F.map_from_entries(
        F.collect_list(
            F.struct("Description", "Counts"))).alias("Description_Counts")
).show(truncate=False)

{'Customer ID': {8548: '3004000304', 8549: '3004002756'},
 'Desc': {8548: {'MAJOR APPLIANCES': 2, 'Home Office' :2, 'ACCESSORIES' : 1, 'OTHER STUFF':2}, 8549: {'MAJOR APPLIANCES': 2, 'ACCESSORIES' :2}}}

import pyspark.sql.functions as f
df = (
    df
    .select('CustomerID', f.explode(f.col('counts')))
    .withColumn('key_separated', f.explode(f.split(f.col('key'), ',')))
    .withColumn('key_separated_trimmed', f.trim(f.col('key_separated')))
    .groupBy('CustomerId', 'key_separated_trimmed')
    .agg(f.sum(f.col('value')).alias('value'))
    .groupBy('CustomerId')
    .agg(f.collect_list(f.col('key_separated_trimmed')).alias('keys'), f.collect_list(f.col('value')).alias('values'))
    .withColumn('counts', f.map_from_arrays(f.col('keys'), f.col('values')))
    .drop('keys', 'values')
)