如何将计算列添加到源 DataFrame

如何解决如何将计算列添加到源 DataFrame

在 SparkR (Databricks) 中，我可以根据 A 列和 C 列的分组值计算 B 列的平均值，如下所示：

library(SparkR)

df <- createDataFrame (
list(list(1L,1,"1",0.1),list(1L,2,0.2),list(3L,3,"3",0.3)),c("a","b","c","d"))

result <- gapplyCollect(
  df,"c"),function(key,x) {
    y <- data.frame(key,mean(x$b),stringsAsFactors = FALSE)
    colnames(y) <- c( "key_a","key_c","mean_b")
    y
  })

这里的源数据框 - df - 用于生成一个新的 - result - 结果 - mean_b 用于 >key_a 和 key_c。这工作正常，但是如何在不创建新数据框的情况下执行相同的操作？以便将 mean_b 作为新列添加到 df？

解决方法

左连接可以使用聚合键将 result$mean_b 作为新列添加到原始 df。

观察以下代码：

library(SparkR)

df <- createDataFrame (
list(list(1L,1,"x",0.1),list(1L,2,0.2),list(3L,3,"y",0.3)),c("a","b","c","d"))

result_schema <- structType(
  structField("key_a","integer"),structField("key_c","string"),structField("mean_b","double"))

result <- gapply(
  df,"c"),function(key,x) {
    y <- data.frame(key,mean(x$b),stringsAsFactors = FALSE)
    colnames(y) <- c("key_a","key_c","mean_b")
    y
  },result_schema)

请注意，我稍微更改了 df 的值，因为您将数字作为字符串。此外，使用 gapply 代替它返回一个 Spark Dataframe，如 df。它还需要生成的 df 的模式。总结：

df：

  a b c   d
1 1 1 x 0.1
2 1 2 x 0.2
3 3 3 y 0.3

result：

  key_a key_c mean_b
1     1     x    1.5
2     3     y    3.0

现在您可以连接两个 Spark 数据帧：

df2 <- join(df,result,(df$a == result$key_a) & (df$c == result$key_c),"left")
collect(drop(df2,c("key_a","key_c")))

可以使用 drop 删除额外的列，并使用 collect 将数据返回给驱动程序：

  a b c   d mean_b
1 1 1 x 0.1    1.5
2 1 2 x 0.2    1.5
3 3 3 y 0.3    3.0