1

I think reading in CSV files are excruciatingly slow using sparklyr. See MVE

library(sparklyr)
library(dplyr)

conf <- spark_config()
conf$spark.executor.memory <- "60GB"
conf$spark.memory.fraction <- 0.9
conf$spark.executor.cores <- 6
conf$spark.dynamicAllocation.enabled <- "false"
sc <- sparklyr::spark_connect(master = "local", config = conf)

library(data.table)

fwrite(data.table(
  id1 = sample(sprintf("id%03d",1:K), N, TRUE),      # large groups (char)
  id2 = sample(sprintf("id%03d",1:K), N, TRUE),      # large groups (char)
  id3 = sample(sprintf("id%010d",1:(N/K)), N, TRUE), # small groups (char)
  id4 = sample(K, N, TRUE),                          # large groups (int)
  id5 = sample(K, N, TRUE),                          # large groups (int)
  id6 = sample(N/K, N, TRUE),                        # small groups (int)
  v1 =  sample(5, N, TRUE),                          # int in range [1,5]
  v2 =  sample(5, N, TRUE),                          # int in range [1,5]
  v3 =  sample(round(runif(100,max=100),4), N, TRUE) # numeric e.g. 23.5749
), "a.csv")

system.time(sparklyr::spark_read_csv(sc, "a", "a.csv"))

I have already tried to increase the level of RAM available to Spark but the reading speed is too slow at 500 seconds! This is crazily slow compared to data.table::fread.

Is there anyway to configure Spark so that it's faster?

3
  • 2
    the file is stored locally? what happens if you fread the CSV & then parallelize it? is your spark context local? just confirming you really have 60Gb local memory... Nov 1, 2018 at 11:48
  • yes everything is local on my computer. I am trying to improve local mode performance
    – xiaodai
    Nov 1, 2018 at 11:49
  • See here for the code used in the dbbenchmark... have you used spark.read.csv as well? Nov 1, 2018 at 11:51

1 Answer 1

1

There are at least three problems here:

  • local mode is not distributed or even parallel. It will just use a single local thread. If you have only one node at your disposal at least try to increase the number of available threads (possibly exceeding the number of available cores).

    In general a single JVM path is not he best approach, especially with largish memory. Even if you don't have multiple nodes at your disposal you can use pseudo-distrubuted with standalone cluster and collocated master and worker.

  • You don't provide schema for the reader, and require schema inference (default value of infer_schema argument). If you want to avoid this overhead you should provide a schema.

  • You eagerly cache data (default value of memory argument), which is both expensive and seldom useful.

Additionally:

  • Such high value of spark.memory.fraction is likely to drive garbage collector crazy filling old gen. Be sure to check GC times, and if there are unusually high, reduce spark.memory.fraction below default (0.6), not increase.

Finally:

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.