java - 自定义 Transformer 中的 Spark (Java) transformSchema()

Question

我想将我的自定义转换器与 StandardScaler 一起使用:

VectorizerTransformer vectorizerTransformer = new VectorizerTransformer(field.getName());
                pipelineStages.add(vectorizerTransformer);
                StandardScaler scaler = new StandardScaler()
                        .setInputCol(vectorizerTransformer.getOutputColumn())
                        .setOutputCol(field.getName() + "_norm")
                        .setWithStd(true)
                        .setWithMean(true);
                pipelineStages.add(scaler);

但是，当我运行时:

PipelineModel pipelineModel = pipeline.fit(dframe);

我遇到了一个异常(exception):

Exception in thread "main" java.lang.IllegalArgumentException: Field "trans_vector" does not exist.
at org.apache.spark.sql.types.StructType$$anonfun$apply$1.apply(StructType.scala:228)
at org.apache.spark.sql.types.StructType$$anonfun$apply$1.apply(StructType.scala:228)
at scala.collection.MapLike$class.getOrElse(MapLike.scala:128)
at scala.collection.AbstractMap.getOrElse(Map.scala:59)
at org.apache.spark.sql.types.StructType.apply(StructType.scala:227)
at org.apache.spark.ml.util.SchemaUtils$.checkColumnType(SchemaUtils.scala:40)
at org.apache.spark.ml.feature.StandardScalerParams$class.validateAndTransformSchema(StandardScaler.scala:68)
at org.apache.spark.ml.feature.StandardScaler.validateAndTransformSchema(StandardScaler.scala:88)
at org.apache.spark.ml.feature.StandardScaler.transformSchema(StandardScaler.scala:124)
at org.apache.spark.ml.Pipeline$$anonfun$transformSchema$4.apply(Pipeline.scala:180)
at org.apache.spark.ml.Pipeline$$anonfun$transformSchema$4.apply(Pipeline.scala:180)
at scala.collection.IndexedSeqOptimized$class.foldl(IndexedSeqOptimized.scala:57)
at scala.collection.IndexedSeqOptimized$class.foldLeft(IndexedSeqOptimized.scala:66)
at scala.collection.mutable.ArrayOps$ofRef.foldLeft(ArrayOps.scala:186)
at org.apache.spark.ml.Pipeline.transformSchema(Pipeline.scala:180)
at org.apache.spark.ml.PipelineStage.transformSchema(Pipeline.scala:70)
at org.apache.spark.ml.Pipeline.fit(Pipeline.scala:132)
at org.sparkexample.PipelineExample.main(PipelineExample.java:90)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:736)
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:185)
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

其中字段的名称是 VectorizerTransformer 的输出字段。

在 VectorizerTransformer 中我有代码:

@Override
public StructType transformSchema(StructType arg0) {
    return arg0;
}

我相信问题出在这里。所以我需要在那里写点东西，但具体是什么？我只是向数据框添加新字段

trans -> trans_vector

Answer 1

@Override
public StructType transformSchema(StructType structType) {
    return structType.add(getOutputColumn(),new VectorUDT(),true);
}

就是这么简单。

注意:我使用了http://supunsetunga.blogspot.ru/2016/05/custom-transformers-for-spark.html作为 java 转换器的代码。