时间戳

首页 > 解决方案 > sparklyr - 在 Apache Spark Join 中包含空值

r - sparklyr - 在 Apache Spark Join 中包含空值

问题描述

在 Apache Spark Join 中包含空值的问题对 Scala、PySpark 和 SparkR 有答案，但对 sparklyr 没有答案。我一直无法弄清楚如何inner_join在 sparklyr 中将连接列中的空值视为相等。有谁知道如何在 sparklyr 中做到这一点？

标签： rapache-sparkjoindplyrsparklyr

解决方案

您可以调用隐式交叉连接：

#' Return a Cartesian product of Spark tables
#'
#' @param df1 tbl_spark
#' @param df2 tbl_spark
#' @param explicit logical If TRUE use crossJoin otherwise 
#'   join without expression
#' @param suffix character suffixes to be used on duplicate names
cross_join <- function(df1, df2, 
    explicit = FALSE, suffix = c("_x", "_y")) {

  common_cols <- intersect(colnames(df1), colnames(df2))

  if(length(common_cols) > 0) {
    df1 <- df1 %>% rename_at(common_cols, funs(paste0(., suffix[1])))
    df2 <- df2 %>% rename_at(common_cols, funs(paste0(., suffix[2])))
  }

  sparklyr::invoke(
    spark_dataframe(df1), 
    if(explicit) "crossJoin" else "join", 
    spark_dataframe(df2)) %>% sdf_register()
}

并过滤结果IS NOT DISTINCT FROM

# Enable Cross joins
sc %>% 
  spark_session() %>% 
  sparklyr::invoke("conf") %>%
  sparklyr::invoke("set", "spark.sql.crossJoin.enabled", "true")

df1 <- copy_to(sc, tibble(id1 = c(NA, "foo", "bar"), val = 1:3))
df2 <- copy_to(sc, tibble(id2 = c(NA, "foo", "baz"), val = 4:6))

df1 %>%
  cross_join(df2) %>% 
  filter(id1 %IS NOT DISTINCT FROM% id2)

# Source: spark<?> [?? x 4]
  id1   val_x id2   val_y
* <chr> <int> <chr> <int>
1 NA        1 NA        4
2 foo       2 foo       5

优化的执行计划：

<jobj[62]>
  org.apache.spark.sql.catalyst.plans.logical.Join
  Join Inner, (id1#10 <=> id2#76)
:- Project [id1#10, val#11 AS val_x#129]
:  +- InMemoryRelation [id1#10, val#11], StorageLevel(disk, memory, deserialized, 1 replicas)
:        +- Scan ExistingRDD[id1#10,val#11]
+- Project [id2#76, val#77 AS val_y#132]
   +- InMemoryRelation [id2#76, val#77], StorageLevel(disk, memory, deserialized, 1 replicas)
         +- Scan ExistingRDD[id2#76,val#77]

<=>操作员应该以同样的方式工作：

df1 %>%
  cross_join(df2) %>% 
  filter(id1 %<=>% id2)

请注意：

  • 隐式交叉连接将失败，如果它后面没有选择将结果提升为散列连接/排序合并连接，或者启用了交叉连接。
  • 在这种情况下不应使用显式交叉连接，因为它将优先于后续选择。

  • 可以使用dplyr样式交叉连接：

    mutate(df1, `_const` = TRUE) %>%  
  inner_join(mutate(df2, `_const` = TRUE), by = c("_const")) %>% 
  select(-`_const`) %>% 
  filter(id1 %IS NOT DISTINCT FROM% id2)

    但我建议不要这样做，因为它不太健壮（取决于上下文优化器可能无法识别出它_const是恒定的）。

推荐阅读