sparklyr - включение нулевых значений в Apache Spark Join - PullRequest
Новая
       4

sparklyr - включение нулевых значений в Apache Spark Join

0 голосов
/ 03 января 2019

Вопрос Включение нулевых значений в Apache Spark Join содержит ответы для Scala, PySpark и SparkR, но не для sparklyr. Мне не удалось понять, как заставить inner_join в sparklyr обрабатывать нулевые значения в столбце соединения как равные. Кто-нибудь знает, как это можно сделать в sparklyr?

1 Ответ

0 голосов
/ 03 января 2019

Вы можете вызвать неявное перекрестное соединение: 

#' 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 является постоянным).

...