使用 Sparklyr 获取 Spark Dataframe 的最大值索引

【问题标题】：Get index of max of Spark Dataframe with Sparklyr使用 Sparklyr 获取 Spark Dataframe 的最大值索引
【发布时间】：2019-03-17 00:26:51
【问题描述】：

我正在尝试从 Spark 数据帧中获取连续最大值的索引。获得最大值是直截了当的。我执行以下操作：

library(sparklyr)
library(dplyr)

config <- spark_config()
sc <- spark_connect(master = "local", config = config)

df <- replicate(n = 3, sample(x = 0:10,size = 10, rep=TRUE)) %>%
  as.data.frame()


sdf <- sdf_copy_to(sc, df, overwrite = T)

sdf %>% spark_apply(
  function(df) {
    return( pmax(df[1], df[2], df[3]) )})

我尝试使用ft_vector_assembler 将这些收集到一个向量中，但我不熟悉返回的数据结构。例如，我无法从以下代码中恢复最大值

sdf %>% ft_vector_assembler(
  input_cols = c("V1", "V2", "V3"), 
  output_col = "features") %>%
  select(features) %>%
  spark_apply( function(df) pmax(df))

感谢任何帮助。

【问题讨论】：

标签： r apache-spark dplyr apache-spark-sql sparklyr

【解决方案1】：

让我们从你的第一个问题开始：

获取最大值很简单。

确实是这样，但是spark_apply 并不是一个可行的方法。相反，最好使用greatest 函数：

sdf %>% mutate(max = greatest(V1, V2, V3))

第二个问题可以使用相同的函数，但是由于sparklyr的限制，你必须直接使用SQL表达式：

expr <- c("V1", "V2", "V3") %>% 
  paste0(
    "CAST(STRUCT(`",
    ., "`, ", seq_along(.),
    ") AS struct<value: double, index: double>)", collapse=", ")  %>% 
  paste0("greatest(", ., ").index AS max_index")

sdf %>% 
  spark_dataframe() %>%
  invoke("selectExpr", list("*", expr)) %>%
  sdf_register()

在 Spark 2.4 中（目前在 sparklyr 中不支持）它可能可能

sdf %>% mutate(max
  max_index = array_max(arrays_zip(array(V1, V2, V3), array(1, 2, 3))).1
)

【讨论】：