Как создать новый массив подстрок из столбца массива строк в фрейме данных искры

У меня есть фрейм данных искры. Один из столбцов представляет собой тип массива, состоящий из массива текстовых строк различной длины. Я ищу способ добавить новый столбец, представляющий собой массив уникальных левых 8 символов этих строк.

df.printSchema()

root
(...)
 |-- arr_agent: array (nullable = true)
 |    |-- element: string (containsNull = true)

пример данных из столбца "arr_agent":

["NRCANL2AXXX", "NRCANL2A"]
["UTRONL2U", "BKRBNL2AXXX", "BKRBNL2A"]
["NRCANL2A"]
["UTRONL2U", "REUWNL2A002", "BKRBNL2A", "REUWNL2A", "REUWNL2N"]
["UTRONL2U", "UTRONL2UXXX", "BKRBNL2A"]
["MQBFDEFFYYY", "MQBFDEFFZZZ", "MQBFDEFF"  ]

Что мне нужно иметь в новой колонке:

["NRCANL2A"]
["UTRONL2U", "BKRBNL2A"]
["NRCANL2A"]
["UTRONL2U", "BKRBNL2A", "REUWNL2A", "REUWNL2N"]
["UTRONL2U", "BKRBNL2A"]
["MQBFDEFF"  ]

Я уже пытался определить udf, который делает это за меня.

from pyspark.sql import functions as F
from pyspark.sql import types as T

def make_list_of_unique_prefixes(text_array, prefix_length=8):
    out_arr = set(t[0:prefix_length] for t in text_array)
    return(out_arr)

make_list_of_unique_prefixes_udf = F.udf(lambda x,y=8: make_list_of_unique_prefixes(x,y), T.ArrayType(T.StringType()))

Но затем звонит:

df.withColumn("arr_prefix8s", F.collect_set( make_list_of_unique_prefixes_udf(F.col("arr_agent") ))) 

Выдает ошибку AnalysisException: grouping expressions sequence is empty,

Любые советы будут оценены. спасибо