pyspark "object has no attribute '_to_java'"

I have a custom transformer that I have created according to this example:
https://stackoverflow.com/questions/32331848/create-a-custom-transformer-in-pyspark-ml

I can use the model.save and my pipeline will save.
Using mleap I get `no attribute to_java`

Here is the transformer:
```
from pyspark import keyword_only ## < 2.0 -> pyspark.ml.util.keyword_only
from pyspark.ml import Transformer
from pyspark.ml.param.shared import HasInputCol, HasOutputCol, Param, Params, TypeConverters
# Available in PySpark >= 2.3.0
from pyspark.ml.util import DefaultParamsReadable, DefaultParamsWritable
from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType, StringType
from selectolax.parser import HTMLParser
class CleanHtml(
Transformer, HasInputCol, HasOutputCol,
# Credits https://stackoverflow.com/a/52467470
# by https://stackoverflow.com/users/234944/benjamin-manns
DefaultParamsReadable, DefaultParamsWritable):
#https://www.google.com/search?q=spark+udf+transformer&oq=spark+udf+transformer&aqs=chrome..69i57.4514j1j7&sourceid=chrome&ie=UTF-8
#stopwords = Param(Params._dummy(), "stopwords", "stopwords",
# typeConverter=TypeConverters.toListString)

@keyword_only
def __init__(self, inputCol=None, outputCol=None):
super(CleanHtml, 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)

# Required in Spark >= 3.0
def setInputCol(self, value):
"""
Sets the value of :py:attr:`inputCol`.
"""
return self._set(inputCol=value)

# Required in Spark >= 3.0
def setOutputCol(self, value):
"""
Sets the value of :py:attr:`outputCol`.
"""
return self._set(outputCol=value)

def _transform(self, dataset):
#todo(aj) hashes function
def clean_hashes(raw):
clean_nonwords = re.compile(r'\S*[^a-zA-Z\s\-\"\']\S*')
cleantext = re.sub(clean_nonwords,'',raw)
return cleantext

def f(raw):

tree = HTMLParser(raw)

if tree.body is None:
return raw

for tag in tree.css('script'):
tag.decompose()
for tag in tree.css('style'):
tag.decompose()

text = tree.body.text(separator='\n')
text = re.sub(r'\n\s*', "\n",text)
text = clean_hashes(text)
print("text: " + text)
return text.strip().strip("\n")

t = StringType()
out_col = self.getOutputCol()
in_col = dataset[self.getInputCol()]
return dataset.withColumn(out_col, udf(f, t)(in_col))
```

here is my pipeline:
```
import re
from pyspark.ml.feature import Tokenizer,StopWordsRemover, CountVectorizer,IDF,StringIndexer
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

#apply extract body remove html
#https://stackoverflow.com/questions/32331848/create-a-custom-transformer-in-pyspark-ml
from pyspark.ml.classification import LogisticRegression

cleanhtml = CleanHtml(inputCol="text", outputCol="clean_text")
tokenizer = Tokenizer(inputCol="clean_text", outputCol="token_text")
stopremove = StopWordsRemover(inputCol='token_text',outputCol='stop_tokens')
count_vec = CountVectorizer(inputCol='stop_tokens',outputCol='c_vec') #TF
idf = IDF(inputCol="c_vec", outputCol="tf_idf") #IDF Scaler
lr = LogisticRegression(regParam=0.01,maxIter=20,featuresCol='features', labelCol='target_int')
from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.linalg import Vector
clean_up = VectorAssembler(inputCols=['tf_idf'],outputCol='features')
pipeline = Pipeline(stages=[cleanhtml,tokenizer,stopremove,count_vec,idf,clean_up,lr])
from pyspark.ml.evaluation import BinaryClassificationEvaluator,MulticlassClassificationEvaluator
paramGrid = ParamGridBuilder()\
.addGrid(lr.regParam, [0.1, 0.001])\
.build()
crossval = CrossValidator(estimator=pipeline,
estimatorParamMaps=paramGrid,
evaluator=MulticlassClassificationEvaluator(labelCol="target_int",metricName='f1'),
numFolds=2)
mdl = crossval.fit(joined)
```

Any help to get the transformer to serialize is appreciated.

Begin with the CleanHtml transformer and the pipeline passed to CrossValidator; reproduce the MLeap serialization failure after model.save succeeds. Compare that path with the PySpark transformer serialization expectations and verify that the fitted pipeline can be exported and loaded without the `_to_java` error.

Written by the indexing model from the issue text.