SparkR groupBy 多列，每个列都应用过滤器

Posted 2023-03-23

【中文标题】SparkR groupBy 多列，每个列都应用过滤器

【英文标题】：SparkR groupBy multiple column with applying filter on each

【发布时间】：2019-12-16 19:39:21

【问题描述】：

我有一个包含超过 5 亿条记录的数据集。我想在多个列上应用 group by 子句来获取计数。在分组时，我还需要确保结果计数仅针对列中的特定值。

我有贷款表，其中有 customer_id,loan_id, installment_amt, installment_status Installment_status 包含多个值 'B'、'N'、'C'

在单个查询中，我想知道每个 customer_id、loan_id、分期付款的总数是多少、只有“B”的分期付款数量和“C”的分期付款数量。

我是 SparkR 的新手，试图做类似下面的事情-

RESULT <- summarize(
  groupBy(LOAN, "customer_id", "loan_id"),
  NO_OF_Installment=count(LOAN$installment_amt),
  BILLED_INSTALLMENTS=count(LOAN$$installment_status=='B'),
  CCANCELLED_INSTALLMENT=count(LOAN$$installment_status=='C')
)

它为 billed_installment 和 cancelled_installment 提供了相同的计数。

我不太确定计数时过滤是否有效。我在the documentation 中没有看到任何内容。但我已经看到这段代码在 R 中工作。

【参考方案1】：

我发现SparkR 代码使用管道更容易阅读，因为它看起来更类似于 Python 或 Scala 版本，所以我将使用 magrittr。

library(magrittr)

基本思想是使用ifelse方法。

在 SparkQL 中：

LOAN %>% createOrReplaceTempView('LOAN')
sql("
select customer_id, loan_id, count(installment_amt) as no_of_installment,
       count(if(installment_status = 'B', 1, NULL)) as billed_installments,
       count(if(installment_status = 'C', 1, NULL)) as cancelled_installments
from loan
group by customer_id, loan_id
") %>% summarize

在“原生”SparkR 中应该是：

LOAN %>% groupBy('customer_id', 'loan_id') %>%
  summarize(
    NO_OF_Installment = count(.$installment_amt),
    BILLED_INSTALLMENTS = count(ifelse(.$installment_status == 'B', 1, NA)),
    CANCELLED_INSTALLMENTS = count(ifelse(.$installment_status == 'C', 1, NA))
  )

我不能 100% 确定您是否需要 NA 或 NULL 作为 ifelse 中的 no 值，但我确实使用 NA 找到了 this 答案。

---

至于为什么您自己的方法不起作用，我认为您的方法适用于 sum 而不是 count。

count 计算一列中非NULL 的行数。 LOAN$installment_status=='C' 是 boolean 列，因此只有在 LOAN$installment_status 是 NULL 时才会是 NULL。 count 不关心列的实际值——它甚至不关心数据类型。

与count 最接近的base R 是length。 length(numeric(100)) 与 length(logical(100)) 相同。

相反，您可能更愿意将其视为sum——base R 等效项类似于sum(installment_status == 'B')。在SparkR 中，这看起来像

sum(as.integer(.$installment_status == 'B'))
# or
sum(ifelse(.$installment_status == 'B', 1, 0))

不幸的是，当base R 隐式地将logical 类型转换为integer 时，sum，SparkR 需要显式转换，因此这两个替代方案使得从boolean 到integer 的转换是显式的。

【讨论】：

谢谢！你的回答帮助我思考了一点不同。而不是 count(if(installment_status = 'B', 1, NULL)) 我想使用 sum(if(installment_status = 'B', 1, 0))。有了这个，我就不用担心空值了