The problem:
------------

I have a column with a list of redundant values, which I need to be converted into a dictionary-like format in a new column of a PySpark dataframe.

The scenario:

Here's my PySpark dataframe:

| A | C | all\_classes |
| --- | --- | --- |
| 10 | RDK | [1, 1, 1, 2, 2] |
| 10 | USW | [1, 2, 2, 2, 2, 2, 2] |
| 8 | RDK | [21, 21, 21, 21, 20] |
| 8 | RDJ | [20, 20, 21] |
| 10 | RDK | [45, 45, 45, 45, 2, 2, 2] |
| 7 | SSW | [6, 6, 6, 19, 19] |

which I created using below line:

```
my_df.groupBy(['A', 'C']).agg(collect_list("Class").alias("all_classes"))

```

Now that I want a new column that will collate data as follows:  

Consider the data of 1'st row from the above column: `[1, 1, 1, 2, 2]`

That needs to be converted into `{1: 3, 2: 2}` which is basically the number itself and no. of times it is duplicated.

My attempt:
-----------

Since, I'm good with python I chose to go with writing a UDF something like this:

```
custom_collect_function = udf(lambda li: {k: li.count(k) for k in set(li)}, StructType(li, ArrayType(elementType=IntegerType()), False))

display(my_df.withColumn("Distribution", custom_collect_function(my_df.all_classes)))

```

I'm clearly failing to set the StructType here and I welcome any other/optimized approach than this. I also welcome any Spark way approach to the same.

The expected output:

| A | C | Distribution | all\_classes |
| --- | --- | --- | --- |
| 10 | RDK | {1: 3, 2: 2} | [1, 1, 1, 2, 2] |
| 10 | USW | {1: 1, 2: 6} | [1, 2, 2, 2, 2, 2, 2] |
| 8 | RDK | {21: 4, 20: 1} | [21, 21, 21, 21, 20] |
| 8 | RDJ | {20: 2, 21: 1} | [20, 20, 21] |
| 10 | RDK | {45: 4, 2: 3} | [45, 45, 45, 45, 2, 2, 2] |
| 7 | SSW | {6: 3, 19: 2} | [6, 6, 6, 19, 19] |