sparklyr 中 R 的 list() 函数的等价物是什么？答案

【问题标题】：What is the equivalent of R's list() function in sparklyr?sparklyr 中 R 的 list() 函数的等价物是什么？
【发布时间】：2018-11-13 09:22:46
【问题描述】：

下面是一个示例 R 代码。我想在 sparklyr 中做同样的事情。

custTrans1 <- Pdt_table %>% 
  group_by(Main_CustomerID) %>% 
  summarise(Invoice = as.vector(list(Invoice_ID)),Industry = as.vector(list(Industry)))

其中 Pdt_table 是 spark 数据框，Main_CustomerID、Invoice_ID 和 Industry 是变量。

我想创建上述变量的列表并将其转换为向量。在sparklyr怎么办？

【问题讨论】：

标签： r apache-spark sparklyr

【解决方案1】：

您可以使用collect_list 或collect_set：

set.seed(1)
df <- copy_to(
  sc, tibble(group = rep(c("a", "b"), 3), value = runif(6)),
  name = "df"
)

result <- df %>% group_by(group) %>% summarise(values = collect_list(value))
result

# Source:   lazy query [?? x 2]
# Database: spark_connection
  group values    
  <chr> <list>    
1 b     <list [3]>
2 a     <list [3]>

which is translated to 以下查询：

result %>% show_query()

<SQL>
SELECT `group`, COLLECT_LIST(`value`) AS `values`
FROM `df`
GROUP BY `group`

对应的execution plan:

result %>% optimizedPlan()

<jobj[213]>
  org.apache.spark.sql.catalyst.plans.logical.Aggregate
  Aggregate [group#259], [group#259, collect_list(value#260, 0, 0) AS values#345]
+- InMemoryRelation [group#259, value#260], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas), `df`
      +- Scan ExistingRDD[group#259,value#260]

和架构（带有array<...> 列）：

root
 |-- group: string (nullable = true)
 |-- values: array (nullable = true)
 |    |-- element: double (containsNull = true)

请记住：

像这样的操作在分布式系统中非常昂贵。
根据数据分布可能不可行。
一般来说，复杂类型在 Spark 中有些难以处理，而 sparklyr 具有整洁的数据焦点，不会让事情变得更容易。为了有效地处理结果，您可能需要一个 Scala 扩展。

【讨论】：