如何将计算列添加到源 DataFrame

Posted 2023-03-23

【中文标题】如何将计算列添加到源 DataFrame

【英文标题】：How to add calculated columns to the source DataFrame

【发布时间】：2021-06-26 21:04:54

【问题描述】：

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

library(SparkR)

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

result <- gapplyCollect(
  df,
  c("a", "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 for key_a 和 key_c。 这工作正常，但是如何在不创建新数据框的情况下执行相同的操作？这样 mean_b 会作为新列添加到 df 吗？

【问题讨论】：

你试过df <- gapplyCollect(而不是result <- gapplyCollect(吗？

@AlexOtt - 是的，它被 colnames 部分中的内容覆盖： colnames(y)

【参考方案1】：

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

观察以下代码：

library(SparkR)

df <- createDataFrame (
list(list(1L, 1, "x", 0.1), list(1L, 2, "x", 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("a", "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 代替它返回像 df 这样的 Spark 数据帧。它还需要生成的 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 Dataframe：

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