I'm having some trouble understanding the creation of custom transformers for Pyspark pipelines.

I am writing a custom transformer that will take the dataframe column Company and remove stray commas:

from pyspark.sql.functions import *

class DFCommaDropper(Transformer):

    def__init__(self, *args, **kwargs):
        self.name = CommaDropper

    def transform(self,df):
        df = df.withColumn('Company', regexp_replace('Company',',','')
        return df

The above code is obviously wrong. I'm unsure what/how to initialize this and then how to use the initialized class instance in the transform function.

Thanks in advance for your help.


1 Answer 1

class StrayCommaRemover(TransformerMixin):
    def __init__(self): //Initialize self by setting some variables here which can be passed as a input to transformer

def fit(self, X, y=None):
    self.columns = X.columns //Setting context based on input Data
    return self

def transform(self, X, y=None): // Actual transformation logic
    X= X.withColumn('Company', regexp_replace('Company',',','')
    return X

You can add above transformer as a step in your pipeline and can call init() and fit() on it.

  • $\begingroup$ What do you mean by 'set context based on input d' $\endgroup$ Commented Jul 10, 2020 at 12:09
  • $\begingroup$ What do you mean by 'set context based on input data'? $\endgroup$ Commented Jul 10, 2020 at 12:11
  • $\begingroup$ Building Transformer context by setting the variables in it to use it further. $\endgroup$ Commented Jul 23, 2020 at 8:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.