4

我对 SPARKR 比较陌生。我下载了 SPARK 1.4 并设置 RStudio 以使用 SPARKR 库。但是我想知道如何将函数应用于分布式 DataFrame 的列中的每个值，有人可以帮忙吗？例如，

这完美地工作

myFunc <- function(x) { paste(x , "_hello")}
c <- c("a", "b", "c")
d <- lapply(c, myFunc)

如何使其适用于分布式 DataFrame。目的是将“_hello”附加到DF的列名称的每个值

DF <- read.df(sqlContext, "TV_Flattened_2.csv", source = "com.databricks.spark.csv", header="true")
SparkR:::lapply(DF$Name, myFunc)

在 SPARK 1.4 发布之前的 alpha 版本的 SPARKR 中似乎已经有了这个能力，为什么现在在 SPARK 1.4 正式版本中缺少这个能力？

4

3 回答 3

2

使用 flatMap，您可以从 DataFrame 中创建一个 RDD，并将该函数应用于所有项目。

c <- c("a", "b", "c")
df <- createDataFrame(sqlContext, as.data.frame(c))
myFunc <- function(x) { paste(x , "_hello")}
d <- flatMap(df, myFunc)
e <- createDataFrame(sqlContext, d)

然而，缺点是仅在 DataFrame 的第一列上执行您期望的操作，它会跳过所有其他列。这可以在以下示例中看到：

c <- c("a", "b", "c")
df <- createDataFrame(sqlContext, as.data.frame(c,u=c(1,2,3)))
myFunc <- function(x) { paste(x , "_hello")}
d <- flatMap(df, myFunc)
e <- createDataFrame(sqlContext, d)

这给出了与第一个示例完全相同的输出，但是 df 从一个额外的列开始。

于 2015-08-17T05:41:13.407 回答
0

我玩了很多次，并没有一个干净的解决方案将函数直接应用于列元素，坦率地说，我不确定这目前是否可行。尽管如此，使用 COLLECT 方法，我们可以执行以下操作：

注意我使用的是 Windows 并输入 powershell

cd D:\Spark\spark-1.4.1-bin-hadoop2.6
./bin/sparkR
c <- c("a", "b", "c")
df <- createDataFrame(sqlContext, as.data.frame(c))
c1 <- collect(df)
myFunc <- function(x) { paste(x , "_hello")}
d <- lapply(c1, myFunc)
df2 <- createDataFrame(sqlContext, as.data.frame(d))
head(df2)

生成您在 R 中打印的内容： 1 a _hello 2 b _hello 3 c _hello

以下是有用的资源：

https://spark.apache.org/docs/latest/api/R/index.html

https://spark.apache.org/docs/latest/sparkr.html

https://databricks.com/blog/2015/08/12/from-pandas-to-apache-sparks-dataframe.html

于 2015-08-17T03:14:26.140 回答
0

Spark 2.x 现在有一个名为 dapply 的函数，它允许您在 SparkR 数据帧的每个分区上运行 R 函数。

来自文档的代码示例：

# Convert waiting time from hours to seconds.
# Note that we can apply UDF to DataFrame.
schema <- structType(structField("eruptions", "double"), structField("waiting", "double"),
                     structField("waiting_secs", "double"))
df1 <- dapply(df, function(x) { x <- cbind(x, x$waiting * 60) }, schema)
head(collect(df1))
##  eruptions waiting waiting_secs
##1     3.600      79         4740
##2     1.800      54         3240
##3     3.333      74         4440
##4     2.283      62         3720
##5     4.533      85         5100
##6     2.883      55         3300

有关更多信息，请参见此处：http: //spark.apache.org/docs/latest/sparkr.html#run-a-given-function-on-a-large-dataset-using-dapply-or-dapplycollect

请注意，如果您使用任何外部 R 库，则需要在工作节点上安装这些库

于 2017-05-21T09:44:09.920 回答