在 sparklyr 中使用外部参数创建和应用 ml_lib 管道

Question

我正在尝试创建和应用一个 Spark ml_pipeline 对象，该对象可以处理会发生变化的外部参数（通常是日期）。根据Spark 文档，这似乎是可能的：参见ParamMap 的部分here

我还没有确切地尝试过如何做到这一点。我在想这样的事情：

table.df <- data.frame("a" = c(1,2,3))
table.sdf <- sdf_copy_to(sc, table.df)

param = 5
param2 = 4

# operation declaration
table2.sdf <- table.sdf %>% 
  mutate(test = param)

# pipeline creation
pipeline_1 = ml_pipeline(sc) %>%
  ft_dplyr_transformer(table2.sdf) %>%
  ml_fit(table.sdf, list("param" = param))

# pipeline application with another value for param
table2.sdf <- pipeline_1 %>% 
  ml_transform(table.sdf, list("param" = param2))

#result

glimpse(table2.sdf %>% select(test))
# doesn work...

Answer 1

这并不是 Spark ML Pipelines 的预期用途。一般而言，将输入数据集转换为适合Pipeline 的格式所需的所有转换都应预先应用，并且仅应将公共组件嵌入为stages。

当使用原生 (Scala) API 时，技术上可以在这种简单的情况下使用空的SQLTransformer：

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.SQLTransformer
import org.apache.spark.ml.param.ParamPair

val df = spark.range(1, 4).toDF("a")

val sqlTransformer = new SQLTransformer()
val pipeline = new Pipeline().setStages(Array(sqlTransformer))

并为fit 提供statement Param

val model = pipeline.fit(
  df,
  ParamPair(sqlTransformer.statement, "SELECT *, 4 AS `test` FROM __THIS__")
)

model.transform(df).show

+---+----+
|  a|test|
+---+----+
|  1|   4|
|  2|   4|
|  3|   4|
+---+----+

和transform:

model.transform(
  df,
  ParamPair(sqlTransformer.statement, "SELECT *, 5 AS `test` FROM __THIS__")
 ).show

+---+----+
|  a|test|
+---+----+
|  1|   5|
|  2|   5|
|  3|   5|
+---+----+

但目前ml_fit 和ml_transform / ml_predict 都不支持额外的Params（您可以看到... 被忽略了）。