Apache Spark：将 Column 作为 Transformer 参数传递答案

【问题标题】：Apache Spark: pass Column as Transformer parameterApache Spark：将 Column 作为 Transformer 参数传递
【发布时间】：2020-02-13 03:45:11
【问题描述】：

我这样定义了一个管道转换器：

class MyTransformer(condition: Column) extends SparkTransformer { 
   override def transform(dataset: Dataset[_]): DataFrame = {...}
 }

然后在管道中使用：

val pipeline = new Pipeline()
pipeline.setStages(Array(new MyTransformer(col("test).equals(lit("value"))))
pipeline.fit(df).transform(mydf)

在我的转换器中，我只想对验证条件的行应用转换。

这会导致序列化问题：

Serialization stack:
- object not serializable (class: org.apache.spark.sql.Column, value: (test = value))
- field (class: my.project.MyTransformer, name: condition, type: class org.apache.spark.sql.Column)
- ...

在我的理解中，Transformer 是被序列化的，以便分发给 executor，所以每个参数都应该是可序列化的。

如何绕过它？有解决办法吗？

谢谢。

【问题讨论】：

标签： apache-spark serialization apache-spark-sql apache-spark-ml

【解决方案1】：

这个问题似乎有点老了... 我不知道我的（未经测试的）想法是否符合您的需求。

解决方案可能是使用 SQL 表达式（字符串实例）

val pipeline = new Pipeline()
pipeline.setStages(Array(new MyTransformer("test = 'value'")))
pipeline.fit(df).transform(mydf)

并在Transformer.transform 方法中使用functions.expr() 将表达式String 转换为Column 实例。

这样，条件是可序列化的，而不可序列化的对象会在需要时在执行器中创建。

【讨论】：