重复调用内存中的 spark dataframe 速度变慢

Question

假设我有 40 个连续的 (DoubleType) 变量，我使用 ft_quantile_discretizer 将它们分为四分位数。识别所有变量的四分位数非常快，因为该函数支持一次执行多个变量。

接下来，我想对这些分桶变量进行一次热编码，但目前不支持通过一次调用对所有这些变量进行一次热编码的功能。因此，我通过循环遍历变量，一次为每个分桶变量输送 ft_string_indexer、ft_one_hot_encoder 和 sdf_separate_column。这样就完成了工作。但是，随着循环的进行，它会大大减慢速度。我认为它运行内存不足，但无法弄清楚如何对此进行编程，以便它在变量之间以相同的速度执行。

如果 q_vars 是连续变量的变量名字符数组（比如 40 个），我怎样才能以更高效的方式编写代码？

for (v in q_vars) {
   data_sprk_q<-data_sprk_q %>% 
       ft_string_indexer(v,paste0(v,"b"),"keep",string_order_type = "alphabetAsc") %>%
       ft_one_hot_encoder(paste0(v,"b"),paste0(v,"bc")) %>%
       sdf_separate_column(paste0(v,"bc"),into=q_vars_cat_list[[v]]) 
}

我还尝试将所有引用的变量作为单个大型管道执行，但这也没有解决问题，所以我认为它与循环本身没有任何关系。

test_text<-paste0("data_sprk_q<-data_sprk_q %>% ", paste0("ft_string_indexer('",q_vars,"',paste0('",q_vars,"','b'),'keep',string_order_type = 'alphabetAsc') %>% ft_one_hot_encoder(paste0('",q_vars,"','b'),paste0('",q_vars,"','bc')) %>% sdf_separate_column(paste0('",q_vars,"','bc'),into=",q_vars_cat_list,")",collapse=" %>% "))
eval(parse(text=test_text))

如有任何帮助，我们将不胜感激。

Answer 1

一般来说，由于 Catalyst 优化器的线性复杂性比线性复杂性更差，因此预计较长的 ML 管道会出现一些（有时是实质性的）减速。除了将流程拆分为多个管道并打破两者之间的沿袭（使用检查点并将数据写入持久存储并将其加载回）之外，目前您无能为力。

然而，您当前的代码在此基础上增加了许多问题：

除非你使用超过10个桶StringIndexer
```
ft_string_indexer(v ,paste0(v, "b"), "keep", string_order_type = "alphabetAsc")
```
只是复制 QuantileDiscretizer 分配的标签。使用字典顺序时，层数越多，行为变得越没用。
可能根本不需要应用 One-Hot-Encoding（在最坏的情况下可能有害），这取决于下游过程，甚至使用线性模型，也可能不是严格的必要的（你可能会争辩说分配的标签是有效的序数，并且记录为标称值，并且增加维度不是期望的结果）。
不过最大的问题是sdf_separate_column的应用。它
- 通过增加表达式的数量来增加计算执行计划的成本。
- 通过将稀疏数据转换为密集数据来增加处理所需的内存量。
- 内部sparklyr在每个索引上使用UserDefinedFunction，有效地导致对同一行的重复分配、解码和垃圾收集给集群带来很大压力。
- 最后但同样重要的是，它丢弃了 Spark ML 广泛使用的列元数据。
我强烈建议不要在此处使用此功能。根据您的评论，您似乎希望在将结果传递给其他算法之前对列进行子集化 - 为此您可以使用 VectorSlicer.

总体而言，您可以将管道重写为

set.seed(1)

df <- copy_to(sc, tibble(x=rnorm(100), y=runif(100), z=rpois(100, 1)))

input_cols <- colnames(df)
discretized_cols <- paste0(input_cols, "_d")
encoded_cols <- paste0(discretized_cols, "_e") %>% setNames(discretized_cols)

discretizer <- ft_quantile_discretizer(
  sc, input_cols = input_cols, output_cols = discretized_cols, num_buckets = 10
)
encoders <- lapply(
  discretized_cols, 
  function(x) ft_one_hot_encoder(sc, input_col=x, output_col=encoded_cols[x])
)

transformed_df <- do.call(ml_pipeline, c(list(discretizer), encoders)) %>%
  ml_fit(df) %>% 
  ml_transform(df)

并在需要时应用 ft_vector_slicer。例如，要从 x 中获取对应于第一、第三和第六个桶的值，您可以：

transformed_df %>% 
  ft_vector_slicer(
    input_col="x_d_e", output_col="x_d_e_s", indices=c(0, 2, 5))

重复调用内存中的 spark dataframe 速度变慢

Slowdown with repeated calls to spark dataframe in memory

r

apache-spark

apache-spark-ml

sparklyr