human
I am working on the file "Cdiscount/Part2-2bis-AIF-PysparkWorkflowPipeline-Cdiscount.ipynb".
The first task is:
Création du pipeline
Création d'un Transformer pour l'étape de stemming.
Dans le calepin précédent, nous avons définie une fonction stemmer à partir de la librairie nltk. Pour que celle-ci puisse être utilisé dans un Pipeline ML, nous devons en faire un objet transformers.
Can you write Python code for it?

What is a the Python code you requested?
assistant
#Here is the Python code you requested.

from pyspark import keyword_only
from pyspark.ml import Transformer
from pyspark.ml.param.shared import HasInputCol, HasOutputCol, Param
from pyspark.sql.functions import udf, col
from pyspark.sql.types import ArrayType, StringType

class MyNltkStemmer(Transformer, HasInputCol, HasOutputCol):

    @keyword_only
    def __init__(self, inputCol=None, outputCol=None):
        super(MyNltkStemmer, self).__init__()
        kwargs = self._input_kwargs
        self.setParams(**kwargs)

    @keyword_only
    def setParams(self, inputCol=None, outputCol=None):
        kwargs = self._input_kwargs
        return self._set(**kwargs)

    def _transform(self, dataset):
        STEMMER = nltk.stem.SnowballStemmer('french')
        def clean_text(tokens):
            tokens_stem = [ STEMMER.stem(token) for token in tokens]
            return tokens_stem
        udfCleanText =  udf(lambda lt : clean_text(lt), ArrayType(StringType()))
        out_col = self.getOutputCol()
        in_col = dataset[self.getInputCol()]
        return dataset.withColumn(out_col, udfCleanText(in_col))