Я действительно новичок в PySpark и пытаюсь перевести некоторый код Python в pyspark. Я начинаю с панды, конвертирую в документ-матрицу терминов, а затем применяю PCA.

UDF:

    class MultiLabelCounter():
        def __init__(self, classes=None):
            self.classes_ = classes

        def fit(self,y):
            self.classes_ = 
    sorted(set(itertools.chain.from_iterable(y)))
            self.mapping = dict(zip(self.classes_,
                                         
    range(len(self.classes_))))
            return self

    def transform(self,y):
        yt = []
        for labels in y:
            data = [0]*len(self.classes_)
            for label in labels:
                data[self.mapping[label]] +=1
            yt.append(data)
        return yt

    def fit_transform(self,y):
        return self.fit(y).transform(y)

    mlb = MultiLabelCounter()

    df_grouped = 
    df_grouped.withColumnRenamed("collect_list(full)","full")

    udf_mlb = udf(lambda x: mlb.fit_transform(x),IntegerType())

    mlb_fitted = df_grouped.withColumn('full',udf_mlb(col("full")))

Я, конечно, получаю NULL результаты.

Я использую искру версии 2.4.4.

РЕДАКТИРОВАТЬ

Добавление образца ввода и вывода по запросу

Вход:

|id|val|
|--|---|
|1|[hello,world]|
|2|[goodbye, world]|
|3|[hello,hello]|

Выход:

|id|hello|goodbye|world|
|--|-----|-------|-----|
|1|1|0|1|
|2|0|1|1|
|3|2|0|0|