Getting a JDBC driver and using it with Spark and sparklyr

Since Spark runs via a JVM, the natural way to establish connections to database systems is using Java Database Connectivity (JDBC). To do that, we will need a JDBC driver which will enable us to interact with the database system of our choice. For this example, we are using MySQL, but we provide details on other RDBMS later in the article.

mkdir -p $HOME/jars
wget -q -t 3 \
  -O $HOME/jars/mysql-connector.zip \
  https://cdn.mysql.com/Downloads/Connector-J/mysql-connector-java-8.0.21.zip 
unzip -q -o \
  -d $HOME/jars \
  $HOME/jars/mysql-connector.zip

jars <- dir("~/jars", pattern = "jar$", recursive = TRUE, full.names = TRUE)
basename(jars)

## [1] "mysql-connector-java-8.0.21.jar"

config <- list(sparklyr.jars.default = jars)
sc <- sparklyr::spark_connect("local", config = config)

Retrieving data from a database with sparklyr

With the Spark connection established, we can connect to our MySQL database from Spark and retrieve the data. {sparklyr} provides a handy spark_read_jdbc() function for this exact purpose. The API maps closely to the Scala API, but it is not very explicit in how to set up the connection. The key here is the options argument to spark_read_jdbc(), which will specify all the connection details we need.

Setting the options argument of spark_read_jdbc()

First, let us create a jdbcConnectionOpts list with the basic connection properties. These are the connection URL and the driver. Below we also explictly specify the user and password, but these can usually also be provided as part of the URL:

# Connection options
jdbcConnectionOpts <- list(
  url = "jdbc:mysql://localhost:3306/testdb",
  driver = "com.mysql.cj.jdbc.Driver",
  user = "rstudio", 
  password = "pass"
)

The last bit of information we need to provide is the identification of the data we want to extract once the connection is established. For this, we can use one of two options:

• dbtable - in case we want to create a Spark DataFrame by extracting contents of a specific table
• query - in case we want to create a Spark DataFrame by executing a SQL query

Loading a specific database table

First let us go with the option to load a database table that we populated with the flights earlier and named test_table, putting it all together and loading the data using spark_read_jdbc():

# Other options specific to the action
jdbcDataOpts <- list(dbtable = "test_table")

# Use spark_read_jdbc() to load the data
test_tbl <- sparklyr::spark_read_jdbc(
  sc = sc,
  name = "test_table",
  options = append(jdbcConnectionOpts, jdbcDataOpts),
  memory = FALSE
)

# Print some records
test_tbl

## # Source: spark<test_table> [?? x 20]
##    row_names  year month   day dep_time sched_dep_time dep_delay arr_time
##    <chr>     <dbl> <dbl> <dbl>    <dbl>          <dbl>     <dbl>    <dbl>
##  1 1          2013     1     1      517            515         2      830
##  2 2          2013     1     1      533            529         4      850
##  3 3          2013     1     1      542            540         2      923
##  4 4          2013     1     1      544            545        -1     1004
##  5 5          2013     1     1      554            600        -6      812
##  6 6          2013     1     1      554            558        -4      740
##  7 7          2013     1     1      555            600        -5      913
##  8 8          2013     1     1      557            600        -3      709
##  9 9          2013     1     1      557            600        -3      838
## 10 10         2013     1     1      558            600        -2      753
## # … with more rows, and 12 more variables: sched_arr_time <dbl>,
## #   arr_delay <dbl>, carrier <chr>, flight <dbl>, tailnum <chr>,
## #   origin <chr>, dest <chr>, air_time <dbl>, distance <dbl>, hour <dbl>,
## #   minute <dbl>, time_hour <chr>

We provided the following arguments:

• sc is the Spark connection that we established using the config that includes necessary jars
• name is a character string with the name to be assigned to the newly generated table within Spark SQL, not the name of the source table we want to read from our database
• options is a list with both the connection options and the data-related options, so we use append() to combine the jdbcConnectionOpts and jdbcDataOpts lists into one
• memory is a logical that tells Spark whether we want to cache the table into memory. A bit more on that and some performance implications below

Executing a query instead

We mentioned above that apart from just loading a table, we can also choose to execute a SQL query and use its result as the source for our Spark DtaFrame. Here is a simple example of that.

# Use `query` instead of `dbtable`
jdbcDataOpts <- list(
  query = "SELECT * FROM test_table WHERE tailnum = 'N14228'"
)

# Use spark_read_jdbc() to load the data
test_qry <- sparklyr::spark_read_jdbc(
  sc = sc,
  name = "test_table",
  options = append(jdbcConnectionOpts, jdbcDataOpts),
  memory = FALSE
)

# Print some records
test_qry

## # Source: spark<test_table> [?? x 20]
##    row_names  year month   day dep_time sched_dep_time dep_delay arr_time
##    <chr>     <dbl> <dbl> <dbl>    <dbl>          <dbl>     <dbl>    <dbl>
##  1 1          2013     1     1      517            515         2      830
##  2 6570       2013     1     8     1435           1440        -5     1717
##  3 7111       2013     1     9      717            700        17      812
##  4 7349       2013     1     9     1143           1144        -1     1425
##  5 10593      2013     1    13      835            824        11     1030
##  6 13775      2013     1    16     1829           1730        59     2117
##  7 18967      2013     1    22     1902           1808        54     2214
##  8 19417      2013     1    23     1050           1056        -6     1143
##  9 19648      2013     1    23     1533           1529         4     1641
## 10 21046      2013     1    25      724            720         4     1000
## # … with more rows, and 12 more variables: sched_arr_time <dbl>,
## #   arr_delay <dbl>, carrier <chr>, flight <dbl>, tailnum <chr>,
## #   origin <chr>, dest <chr>, air_time <dbl>, distance <dbl>, hour <dbl>,
## #   minute <dbl>, time_hour <chr>

Note that the only element that changed is the jdbcDataOpts list, which now contains a query element instead of a dbtable element.

Oracle

hostName <- "0.0.0.0"
portNumber <- "1521"
serviceName <- "service_name"

jdbcConnectionOpts <- list(
  user = "username",
  password = "password",
  driver = "oracle.jdbc.driver.OracleDriver",
  fetchsize = "100000",
  url = paste0(
    "jdbc:oracle:thin:@//",
    hostName, ":", portNumber,
    "/", serviceName
  )
)

serviceName <- "service_name"

jdbcConnectionOpts <- list(
  user = "username",
  password = "password",
  driver = "oracle.jdbc.driver.OracleDriver",
  fetchsize = "100000",
  url = paste0("jdbc:oracle:thin:@", serviceName)
)

MS SQL Server

serverName <- "0.0.0.0"
portNumber  <- "1433"
databaseName <- "db_name"

jdbcConnectionOpts <- list(
  user = "username",
  password = "password",
  driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver",
  fetchsize = "100000",
  url = paste0(
    "jdbc:sqlserver://",
    serverName, ":", portNumber,
    ";databaseName=", databaseName
  )
)

Even more RDBM Systems

Logos of R, sparklyr, Spark and selected RDBMS systems

Vlad Mihalcea wrote a very useful article on JDBC Driver Connection URL strings which has the connection URL details for several other common database systems.

The memory argument

The memory argument to spark_read_jdbc() can prove very important when performance is of interest. What happens when using the default memory = TRUE is that the table in the Spark SQL context is cached using CACHE TABLE and a SELECT count(*) FROM query is executed on the cached table. This forces Spark to perform the action of loading the entire table into memory.

Depending on our use case, it might be much more beneficial to use memory = FALSE and only cache into Spark memory the parts of the table (or processed results) that we need, as the most time-costly operations usually are data transfers over the network. Transferring as little data as possible from the database into Spark memory may bring significant performance benefits.

This is a bit difficult to show with our toy example, as everything is physically happening inside the same container (and therefore the same file system), but differences can be observed even with this setup and our small dataset:

microbenchmark::microbenchmark(
  times = 10,
  setup = {
    library(dplyr)
    library(sparklyr)
    sparklyr::spark_disconnect_all()
    sc <- sparklyr::spark_connect("local", config = config)
  },
  
  # with memory=TRUE (the default)
  eager = {
    one <- sparklyr::spark_read_jdbc(
      sc = sc,
      name = "test",
      options = append(jdbcConnectionOpts, list(dbtable = "test_table"))
    ) %>%
      filter(tailnum == "N14228") %>%
      select(tailnum, distance) %>%
      compute("test")
  },

  # with memory=FALSE
  lazy = {
    two <- sparklyr::spark_read_jdbc(
      sc = sc,
      name = "test",
      options = append(jdbcConnectionOpts, list(dbtable = "test_table")),
      memory = FALSE
    ) %>% 
      filter(tailnum == "N14228") %>%
      select(tailnum, distance) %>%
      compute("test")
  }
)

# Unit: seconds
#  expr       min       lq     mean   median       uq      max neval
# eager 15.460844 16.24838 17.07560 17.03592 17.88299 18.73005    10
#  lazy  9.821039 10.12435 10.40718 10.42766 10.70024 10.97283    10

We see that the “lazy” approach that does not cache the entire table into memory has yielded the result around 41% faster. This is of course by no means a relevant benchmark for real-life data loads but can provide some insight into optimizing the loads.

Partitioning

Partitioning the data can bring a very significant performance boost and we will look into setting it up and optimizing it in detail in a separate article.