Ggtcf

For each categorical variable in dataset, I want to get counts and summary stats for each level. I can do this using dlookr R package using their diagnose_category() function. Since at work I don't have that package I recreated the function using dplyr.

In sparklye I am able to get counts for a single variable at a time. Need help to extend it all categorical variable.

Need Help:

Implement the function via SparklyR

Table 1: Final output needed:

# A tibble: 20 x 6
 variables levels N freq ratio rank
 <chr> <ord> <int> <int> <dbl> <int>
 1 cut Ideal 53940 21551 40.0 1
 2 cut Premium 53940 13791 25.6 2
 3 cut Very Good 53940 12082 22.4 3
 4 cut Good 53940 4906 9.10 4
 5 cut Fair 53940 1610 2.98 5
 6 color G 53940 11292 20.9 1
 7 color E 53940 9797 18.2 2
 8 color F 53940 9542 17.7 3
 9 color H 53940 8304 15.4 4
10 color D 53940 6775 12.6 5
11 color I 53940 5422 10.1 6
12 color J 53940 2808 5.21 7
13 clarity SI1 53940 13065 24.2 1
14 clarity VS2 53940 12258 22.7 2
15 clarity SI2 53940 9194 17.0 3
16 clarity VS1 53940 8171 15.1 4
17 clarity VVS2 53940 5066 9.39 5
18 clarity VVS1 53940 3655 6.78 6
19 clarity IF 53940 1790 3.32 7
20 clarity I1 53940 741 1.37 8

R Code:

# Categorical Variable Profile
# Table based on dlookr package, diagnose_category() function
# variables : variable names
# types: the data type of the variable
# levels: level names
# N : Number of observation
# freq : Number of observation at the level
# ratio : Percentage of observation at the level
# rank : Rank of occupancy ratio of levels 

library(ggplot2)
library(dplyr)
library(tidyr)
library(purrr)
library(tibble)
library(stringr)

# Helper Function
cat_level_summary <- function(df,x) 
 count(df,x, sort = TRUE) %>% 
 transmute(levels = x, N = sum(n), freq = n,
 ratio = n / sum(n) * 100, rank = row_number())
 

# Loading
diamonds_tbl <- diamonds

# Main Code
CategoricalVariableProfile <- diamonds_tbl %>%
 select_if(~!is.numeric(.)) %>% 
 map(~cat_level_summary(data.frame(x=.x), x)) %>%
 do.call(rbind.data.frame, .) %>%
 rownames_to_column(., "variables")%>%
 mutate(variables = str_match(variables, ".*(?=\.)")[, 1] )

Spark Code:

#Spark data Table
diamonds_tbl <- copy_to(sc, diamonds, "diamonds", overwrite = TRUE)

CategoricalVariableProfile <- diamonds_tbl %>% 
 group_by(cut) %>%
 summarize(count = n()) %>%
 sdf_register("CategoricalVariableProfile")

Flatten your data using sdf_gather:

long <- diamonds_tbl %>% 
 select(cut, color, clarity) %>% 
 sdf_gather("variable", "level", "cut", "color", "clarity")

Aggregate by variable and level:

counts <- long %>% group_by(variable, level) %>% summarise(freq = n())

And finally apply required window functions:

result <- counts %>%
 arrange(-freq) %>% 
 mutate(
 rank = rank(),
 total = sum(freq, na.rm = TRUE),
 ratio = freq / total * 100)

Which will give you

result

# Source: spark<?> [?? x 6]
# Groups: variable
# Ordered by: -freq
 variable level freq rank total ratio
 <chr> <chr> <dbl> <int> <dbl> <dbl>
 1 cut Ideal 21551 1 53940 40.0 
 2 cut Premium 13791 2 53940 25.6 
 3 cut Very Good 12082 3 53940 22.4 
 4 cut Good 4906 4 53940 9.10
 5 cut Fair 1610 5 53940 2.98
 6 clarity SI1 13065 1 53940 24.2 
 7 clarity VS2 12258 2 53940 22.7 
 8 clarity SI2 9194 3 53940 17.0 
 9 clarity VS1 8171 4 53940 15.1 
10 clarity VVS2 5066 5 53940 9.39
# … with more rows

with following optimized plan

optimizedPlan(result)

<jobj[165]>
 org.apache.spark.sql.catalyst.plans.logical.Project
 Project [variable#524, level#525, freq#1478L, rank#1479, total#1480L, ((cast(freq#1478L as double) / cast(total#1480L as double)) * 100.0) AS ratio#1481]
+- Window [rank(_w1#1493L) windowspecdefinition(variable#524, _w1#1493L ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rank#1479], [variable#524], [_w1#1493L ASC NULLS FIRST]
 +- Window [sum(freq#1478L) windowspecdefinition(variable#524, specifiedwindowframe(RowFrame, unboundedpreceding$(), unboundedfollowing$())) AS total#1480L], [variable#524]
 +- Project [variable#524, level#525, freq#1478L, -freq#1478L AS _w1#1493L]
 +- Sort [-freq#1478L ASC NULLS FIRST], true
 +- Aggregate [variable#524, level#525], [variable#524, level#525, count(1) AS freq#1478L]
 +- Generate explode(map(cut, cut#19, color, color#20, clarity, clarity#21)), [0, 1, 2], false, [variable#524, level#525]
 +- Project [cut#19, color#20, clarity#21]
 +- InMemoryRelation [carat#18, cut#19, color#20, clarity#21, depth#22, table#23, price#24, x#25, y#26, z#27], StorageLevel(disk, memory, deserialized, 1 replicas)
 +- Scan ExistingRDD[carat#18,cut#19,color#20,clarity#21,depth#22,table#23,price#24,x#25,y#26,z#27]

and query (sdf_gather component not included):

dbplyr::remote_query(result)

<SQL> SELECT `variable`, `level`, `freq`, `rank`, `total`, `freq` / `total` * 100.0 AS `ratio`
FROM (SELECT `variable`, `level`, `freq`, rank() OVER (PARTITION BY `variable` ORDER BY -`freq`) AS `rank`, sum(`freq`) OVER (PARTITION BY `variable`) AS `total`
FROM (SELECT *
FROM (SELECT `variable`, `level`, count(*) AS `freq`
FROM `sparklyr_tmp_ded2576b9f1`
GROUP BY `variable`, `level`) `dsbksdfhtf`
ORDER BY -`freq`) `obyrzsxeus`) `ekejqyjrfz`