Чтение кастомного трансформера pyspark - PullRequest
Новая
       27

Чтение кастомного трансформера pyspark

0 голосов
/ 21 сентября 2018

Поработав с этим в течение долгого времени, в Spark 2.3 я наконец смог сохранить чистый преобразователь на python.Но я получаю сообщение об ошибке при загрузке преобразователя обратно.

Я проверил содержимое сохраненного файла и нашел все соответствующие переменные, сохраненные в файле в HDFS.Было бы замечательно, если бы кто-то заметил, чего мне не хватает в этом простом преобразователе. 

from pyspark.ml import Transformer
from pyspark.ml.param.shared import Param,Params,TypeConverters

class AggregateTransformer(Transformer,DefaultParamsWritable,DefaultParamsReadable):
    aggCols = Param(Params._dummy(), "aggCols", "",TypeConverters.toListString)
    valCols = Param(Params._dummy(), "valCols", "",TypeConverters.toListString)

    def __init__(self,aggCols,valCols):
        super(AggregateTransformer, self).__init__()
        self._setDefault(aggCols=[''])
        self._set(aggCols = aggCols)
        self._setDefault(valCols=[''])
        self._set(valCols = valCols)

    def getAggCols(self):
        return self.getOrDefault(self.aggCols)

    def setAggCols(self, aggCols):
        self._set(aggCols=aggCols)

    def getValCols(self):
        return self.getOrDefault(self.valCols)

    def setValCols(self, valCols):
        self._set(valCols=valCols)

    def _transform(self, dataset):
        aggFuncs = []
        for valCol in self.getValCols():
            aggFuncs.append(F.sum(valCol).alias("sum_"+valCol))
            aggFuncs.append(F.min(valCol).alias("min_"+valCol))
            aggFuncs.append(F.max(valCol).alias("max_"+valCol))
            aggFuncs.append(F.count(valCol).alias("cnt_"+valCol))
            aggFuncs.append(F.avg(valCol).alias("avg_"+valCol))
            aggFuncs.append(F.stddev(valCol).alias("stddev_"+valCol))

        dataset = dataset.groupBy(self.getAggCols()).agg(*aggFuncs)
        return dataset

Я получаю эту ошибку, когда загружаю экземпляр этого преобразователя после его сохранения. 

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-172-44e20f7e3842> in <module>()
----> 1 x = agg.load("/tmp/test")

/usr/hdp/current/spark2.3-client/python/pyspark/ml/util.py in load(cls, path)
    309     def load(cls, path):
    310         """Reads an ML instance from the input path, a shortcut of `read().load(path)`."""
--> 311         return cls.read().load(path)
    312 
    313 

/usr/hdp/current/spark2.3-client/python/pyspark/ml/util.py in load(self, path)
    482         metadata = DefaultParamsReader.loadMetadata(path, self.sc)
    483         py_type = DefaultParamsReader.__get_class(metadata['class'])
--> 484         instance = py_type()
    485         instance._resetUid(metadata['uid'])
    486         DefaultParamsReader.getAndSetParams(instance, metadata)

TypeError: __init__() missing 2 required positional arguments: 'aggCols' and 'valCols'

1 Ответ

0 голосов
/ 21 сентября 2018

Разобрался с ответом!

Проблема заключалась в том, что читатель инициализировал новый класс Transformer, но функция init для моего AggregateTransformer не имела значений по умолчанию для аргументов.

Итакизменение следующей строки кода решило проблему! 

def __init__(self,aggCols=[],valCols=[]):

Оставив здесь ответ и вопрос, поскольку мне было невероятно трудно найти рабочий пример чистого преобразователя Python, который можно было бы сохранить и прочитатьвернуться куда угодно!Это может помочь кому-то, ищущему это.

...