Скажем, у меня есть несколько объектов / столбцов в кадре данных, к которым я применяю обычный OneHotEncoder, и один (пусть, n-й) столбец, к которому мне нужно применить свой пользовательский OneHotEncoder. Затем мне нужно использовать VectorAssembler, чтобы собрать эти функции, и поместить в конвейер, наконец подгоняя мои trainData и получая прогнозы из моих testData, например:
val sIndexer1 = new StringIndexer().setInputCol("my_feature1").setOutputCol("indexed_feature1")
// ... let, n-1 such sIndexers for n-1 features
val featureEncoder = new OneHotEncoderEstimator().setInputCols(Array(sIndexer1.getOutputCol), ...).
setOutputCols(Array("encoded_feature1", ... ))
// **need to insert output from my custom OneHotEncoder function (please see below)**
// (which takes the n-th feature as input) in a way that matches the VectorAssembler below
val vectorAssembler = new VectorAssembler().setInputCols(featureEncoder.getOutputCols + ???).
setOutputCol("assembled_features")
...
val pipeline = new Pipeline().setStages(Array(sIndexer1, ...,featureEncoder, vectorAssembler, myClassifier))
val model = pipeline.fit(trainData)
val predictions = model.transform(testData)
Как я могу изменить построение vectorAssembler, чтобы он мог принимать выходные данные из пользовательского OneHotEncoder? Проблема в том, что мой желаемый oheEncodingTopN () не может / не должен ссылаться на «фактический» фрейм данных , поскольку он будет частью конвейер (для применения к trainData / testData).
Примечание:
Я проверил, что пользовательский OneHotEncoder (см. ссылка ) работает, как и ожидалось, отдельно, например, на trainData. По сути, oheEncodingTopN применяет OneHotEncoding к входному столбцу, но только для верхних N частых значений (например, N = 50) и помещает все остальные нечастые значения в фиктивный столбец (скажем, «по умолчанию»), например:
val oheEncoded = oheEncodingTopN(df, "my_featureN", 50)
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.{col, lit, when}
import org.apache.spark.sql.Column
def flip(col: Column): Column = when(col === 1, lit(0)).otherwise(lit(1))
def oheEncodingTopN(df: DataFrame, colName: String, n: Int): DataFrame = {
df.createOrReplaceTempView("data")
val topNDF = spark.sql(s"select $colName, count(*) as count from data group by $colName order by count desc limit $n")
val pivotTopNDF = topNDF.
groupBy(colName).
pivot(colName).
count().
withColumn("default", lit(1))
val joinedTopNDF = df.join(pivotTopNDF, Seq(colName), "left").drop(colName)
val oheEncodedDF = joinedTopNDF.
na.fill(0, joinedTopNDF.columns).
withColumn("default", flip(col("default")))
oheEncodedDF
}