Calling spark window functions in R using sparklyr

Question

I have been trying to replicate the following pyspark snippet in sparklyr but no luck.

from pyspark.sql.window import Window
from pyspark.sql.functions import concat, col, lit, approx_count_distinct, countDistinct

df = spark.sql("select * from mtcars")

dff = df.withColumn("test", concat(col("gear"), lit(" "), col("carb")))
w = Window.partitionBy("cyl").orderBy("cyl")
  
dff.withColumn("distinct", approx_count_distinct("test").over(w)).show()

The concatenate bit I did manage to get to work like so:

tbl(sc, "mtcars")%>% 
  spark_dataframe() %>% 
  invoke("withColumn", 
         "concat", 
         invoke_static(sc, "org.apache.spark.sql.functions", "expr", "concat(gear, carb)")) %>% 
  sdf_register()

I can't seem to figure out how to invoke the Window.partitionBy() and Window.orderBy()

# Doesn't work
w <- invoke_static(sc, "org.apache.spark.sql.expressions.Window", "partitionBy", "cyl")

Some pointers would be help a lot !

score 1 · Answer 1 · answered Sep 22 '20 at 12:46

1

This should get you going:

w <- orderBy(windowPartitionBy("cyl"), "cyl")
dff <- select(dff, over(approx_count_distinct("test"), w))

answered Sep 22 '20 at 12:46

jayrythium

679
4
11

Thank you for your response! Unfortunately, I can only use `sparklyr` and not `SparkR`. Any ideas on how to do this using `sparklyr` ? – royr2 Sep 22 '20 at 16:33
I apologize, I don't have any experience with sparklyr. Good luck! – jayrythium Sep 22 '20 at 16:50

edog429 · Accepted Answer · 2020-09-24T02:35:00.697

You can pipe the sql directly.

mtcars_spk <- copy_to(sc, mtcars,"mtcars_spk",overwrite = TRUE)
mtcars_spk2 <- mtcars_spk %>%
                dplyr::mutate(test = paste0(gear, " ",carb)) %>%
                dplyr::mutate(discnt = sql("approx_count_distinct(test) OVER (PARTITION BY cyl)"))

It is worth noting here that this is a rare case and other window functions are supported in sparklyr. If you wanted just the count or a min(gear) partitioned by cyl you could do that easily.

mtcars_spk <- copy_to(sc, mtcars,"mtcars_spk",overwrite = TRUE)
mtcars_spk <- mtcars_spk %>%
                group_by(cyl) %>%
                arrange(cyl) %>%
                mutate(cnt = count()
                       ,mindis= min(disp)

Linking in similar threads:

Calling spark window functions in R using sparklyr

2 Answers2

Linked