Spark & R: data frame operations with SparkR

In this third tutorial (see the previous one) we will introduce more advanced concepts about SparkSQL with R that you can find in the SparkR documentation, applied to the 2013 American Community Survey housing data. These concepts are related with data frame manipulation, including data slicing, summary statistics, and aggregations. We will use them in combination with ggplot2 visualisations. We will explain what we do at every step but, if you want to go deeper into ggplot2 for exploratory data analysis, I did this Udacity on-line course in the past and I highly recommend it!

All the code for these series of Spark and R tutorials can be found in its own GitHub repository. Go there and make it yours.

Creating a SparkSQL Context and Loading Data

In order to explore our data, we first need to load it into a SparkSQL data frame. But first we need to init a SparkSQL context. The first thing we need to do is to set up some environment variables and library paths as follows. Remember to replace the value assigned to SPARK_HOME with your Spark home folder.

# Set Spark home and R libs
Sys.setenv(SPARK_HOME='/home/cluster/spark-1.5.0-bin-hadoop2.6')
.libPaths(c(file.path(Sys.getenv('SPARK_HOME'), 'R', 'lib'), .libPaths()))

Now we can load the SparkR library as follows.

library(SparkR)

    Attaching package: ‘SparkR’
    
    The following objects are masked from ‘package:stats’:
    
        filter, na.omit
    
    The following objects are masked from ‘package:base’:
    
        intersect, rbind, sample, subset, summary, table, transform

And now we can initialise the Spark context as in the official documentation. In our case we are use a standalone Spark cluster with one master and seven workers. If you are running Spark in local node, use just master='local'. Additionally, we require a Spark package from Databricks to read CSV files (more on this in the previous notebook).

sc <- sparkR.init(master='spark://169.254.206.2:7077', sparkPackages="com.databricks:spark-csv_2.11:1.2.0")

    Launching java with spark-submit command /home/cluster/spark-1.5.0-bin-hadoop2.6/bin/spark-submit  --packages com.databricks:spark-csv_2.11:1.2.0 sparkr-shell /tmp/RtmpfRY7gu/backend_port4c2413c05644 

And finally we can start the SparkSQL context as follows.

sqlContext <- sparkRSQL.init(sc)

Now that we have our SparkSQL context ready, we can use it to load our CSV data into data frames. We have downloaded our 2013 American Community Survey dataset files in notebook 0, so they should be stored locally. Remember to set the right path for your data files in the first line, ours is /nfs/data/2013-acs/ss13husa.csv.

housing_a_file_path <- file.path('', 'nfs','data','2013-acs','ss13husa.csv')
housing_b_file_path <- file.path('', 'nfs','data','2013-acs','ss13husb.csv')

Now let's read into a SparkSQL dataframe. We need to pass four parameters in addition to the sqlContext:

• The file path.
• header='true' since our csv files have a header with the column names.
• Indicate that we want the library to infer the schema.
• And the source type (the Databricks package in this case).

And we have two separate files for both, housing and population data. We need to join them.

housing_a_df <- read.df(sqlContext, 
                            housing_a_file_path, 
                            header='true', 
                            source = "com.databricks.spark.csv", 
                            inferSchema='true')

housing_b_df <- read.df(sqlContext, 
                            housing_b_file_path, 
                            header='true', 
                            source = "com.databricks.spark.csv", 
                            inferSchema='true')

housing_df <- rbind(housing_a_df, housing_b_df)

Let's check that we have everything there by counting the files and listing a few of them.

nrow(housing_df)

1476313

head(housing_df)

Giving ggplot2 a try

Before we dive into data selection and aggregations, let's try plotting something using ggplot2. We will use this library all the time during our exploratory data analysis, and we better mke sure how to use it with SparkSQL results.

# if it isn't installed, 
# install install.packages("ggplot2") 
# from the R console, specifying a CRAN mirror
    
library(ggplot2)

What if we directly try to use our SparkSQL DataFrame class into a ggplot?

c <- ggplot(data=housing_df, aes(x=factor(REGION)))

    Error: ggplot2 doesn't know how to deal with data of class DataFrame

Obviously it doesn't work that way. The ggplot function doesn't know how to deal with that type of distributed data frames (the Spark ones). Instead, we need to collect the data locally as follows.

housing_region_df_local <- collect(select(housing_df,"REGION"))

Let's have a look at what we got.

str(housing_region_df_local)

    'data.frame':	1476313 obs. of  1 variable:
     $ REGION: int  3 3 3 3 3 3 3 3 3 3 ...

That is, when we collect results from a SparkSQL DataFrame we get a regular R data.frame. Very convenient since we can manipulate it as we need to. For example, let's convert that int values we have for REGION to a factor with the proper names. From our data dictionary we will get the meaning of the REGION variable, as well as the different values it can take.

housing_region_df_local$REGION <- factor(
        x=housing_region_df_local$REGION, 
        levels=c(1,2,3,4,9),
        labels=c('Northeast', 'Midwest','South','West','Puerto Rico')
)

And now we are ready to create the ggplot object as follows.

c <- ggplot(data=housing_region_df_local, aes(x=factor(REGION)))

And now we can give the plot a proper representation (e.g. a bar plot).

c + geom_bar() + xlab("Region")

We will always follow the same approach. First we make some sort of oepration with the SparkSQL DataFrame object (e.g. a selection), then we collect results, and then we prepare the resulting data.frame to be represented using ggplot2. But think about the previous. We just represented all the samples for a given column. That is almost a million and a half data points, and we are pushing our local R environment and ggplot2 a lot. In the case of the bar plot we didn't really experience any problems, cause is sort of aggregating data inside. But we will struggle to do scatter plots this way. The preferred kind of visualisations will be those that come from data from aggreations on SparkSQL DataFrames as we will see in further sections.

Data Selection

In this section we will demonstrate how to select data from a SparkSQL DataFrame object using SparkR.

select and selectExpr

We already made use of the select function. but let's have a look at the documentation.

?select

We see there that have two flavours of select. One that gets a list of column names (this is the one we used so far) and another one called selectExpr that we pass a string containing a SQL expression.

Of course we can pass more than a column name.

collect(select(housing_df, "REGION", "VALP"))

When passing column names we can also use the R notation data.frame$column.name so familiar to R users. How does this notation compares to the name-based one in terms of performance?

system.time(
        collect(select(housing_df, housing_df$VALP))
)

       user  system elapsed 
     30.086   0.032  48.046 

system.time(
        collect(select(housing_df, "VALP"))
)

       user  system elapsed 
     28.766   0.012  46.358 

When using the $ notation, we can even pass expressions as follows.

head(select(housing_df, housing_df$VALP / 100))

So what's the point of selectExpr then? Well, we can pass more complex SQL expressions. For example.

head(selectExpr(housing_df, "(VALP / 100) as VALP_by_100"))

filter, subset, and sql

The previous functions allow us selecting columns. In order to select rows, we will use filter and contains. Call the docs as follows if you want to know more about it.

?filter

With filter we filter the rows of a DataFrame according to a given condition that we pass as argument. We can define conditions as SQL conditions using column names or by using the $ notation.

For example, and following with our property values column, let's select property values higher than 1000 for the south region.

system.time(
        housing_valp_1000 <- collect(filter(select(housing_df, "REGION", "VALP"), "VALP > 1000"))
)

       user  system elapsed 
     38.043   0.184  56.259 

housing_valp_1000

Take into account that we can also perform the previous selection and filtering by using SQL queries agains the SparkSQL DataFrame. In order to do that we need to register the table as follows.

registerTempTable(housing_df, "housing")

And then we can use SparkR sql function using the sqlContext as follows.

system.time(
        housing_valp_1000_sql <- collect(sql(sqlContext, "SELECT REGION, VALP FROM housing WHERE VALP >= 1000"))
)

       user  system elapsed 
     38.862   0.008  56.747 

housing_valp_1000_sql

This last method might result more clear and flexible when we need to perform complex queries with multiple conditions. Using filter and select combinations might get verbose versus the clarity of the SQL lingua franca.

But there is another way of subsetting data frames in a functional way. A way that is very familiar to R users. It is by using the function subset. Just have a look at the help page.

?subset

And we use it as follows.

system.time(
        housing_valp_1000_subset <- collect(subset(
            housing_df, housing_df$VALP>1000, 
            c("REGION","VALP")
        ))
)

       user  system elapsed 
     39.751   0.020  57.425 

housing_valp_1000_subset

Even more, we can use the [] notation we use with R data.frame objects with SparkSQL DataFrames thanks to SparkR. For example.

system.time(
        housing_valp_1000_bracket <- collect(
            housing_df[housing_df$VALP>1000, c("REGION","VALP")]
        )
)

       user  system elapsed 
     39.090   0.013  56.381 

housing_valp_1000_bracket

That is, we have up to four different ways of subsetting a data frame with SparkR. We can plot any of the previous resulting data frames with a ggplot2 chart as we did before.

housing_valp_1000_bracket$REGION <- factor(
        x=housing_valp_1000_bracket$REGION, 
        levels=c(1,2,3,4,9),
        labels=c('Northeast', 'Midwest','South','West','Puerto Rico')
)
c <- ggplot(data=housing_region_df_local, aes(x=factor(REGION)))
c + geom_bar() + ggtitle("Samples with VALP>1000") + xlab("Region")

Finally, a function that is useful, specially when imputing missing values in data frames is isNaN that can be applied to columns as we do with regular R data frames.

Data Aggregation and Sorting

In the previous notebook we already had a look at summary/describe that we can pass column names and get summary statistics that way. If we want instead to be specific about the statistic we want, SparkR also defines the following aggregation functions that we can apply to DataFrame objects columns:

• avg
• min
• max
• sum
• countDistinct
• sumDistinct

We use them passing columns with the $ notation, and they return columns, so they need to be part of a select call for a DataFrame. For example.

collect(select(housing_df, avg(housing_df$VALP)))

groupBy and summarize / agg

A basic operation when doing data aggregations on data frames is groupBy. Basically it groups the DataFrame we pass using the specified columns, so we can run aggregation on them. We use it in combination with summarize/agg in order to apply aggregation functions. For example, using the previous avg example, let's averagle property values by region as follows.

housing_avg_valp <- collect(agg(
        groupBy(housing_df, "REGION"), 
        NUM_PROPERTIES=n(housing_df$REGION),
        AVG_VALP = avg(housing_df$VALP), 
        MAX_VALUE=max(housing_df$VALP),
        MIN_VALUE=min(housing_df$VALP)
))
housing_avg_valp$REGION <- factor(
        housing_avg_valp$REGION, 
        levels=c(1,2,3,4,9), 
        labels=c('Northeast', 'Midwest','South','West','Puerto Rico')
)
housing_avg_valp

We can add as many summary/aggregation columns as functions we want to calculate. There is also the possibility of adding several levels of grouping. For example, let's add the number of bedrooms (BDSP in our dictionary) as follows.

housing_avg_valp <- collect(agg(
        groupBy(housing_df, "REGION", "BDSP"), 
        NUM_PROPERTIES=n(housing_df$REGION),
        AVG_VALP = avg(housing_df$VALP), 
        MAX_VALUE=max(housing_df$VALP),
        MIN_VALUE=min(housing_df$VALP)
))
housing_avg_valp$REGION <- factor(
        housing_avg_valp$REGION, 
        levels=c(1,2,3,4,9), 
        labels=c('Northeast', 'Midwest','South','West','Puerto Rico')
)
housing_avg_valp

arrange

One last thing. We can arrange a DataFrame as follows.

head(arrange(select(housing_df, "REGION", "VALP"), desc(housing_df$VALP)))

Or we can arrange the result of our aggregations.

housing_avg_agg <- agg(
            groupBy(housing_df, "REGION", "BDSP"), 
            NUM_PROPERTIES=n(housing_df$REGION),
            AVG_VALP = avg(housing_df$VALP), 
            MAX_VALUE=max(housing_df$VALP),
            MIN_VALUE=min(housing_df$VALP)
)
housing_avg_sorted <- head(arrange(
           housing_avg_agg,   
           desc(housing_avg_agg$AVG_VALP)
))
    
housing_avg_sorted$REGION <- factor(
        housing_avg_sorted$REGION, 
        levels=c(1,2,3,4,9), 
        labels=c('Northeast', 'Midwest','South','West','Puerto Rico')
)
housing_avg_sorted

Conclusions

So that's it. In the next tutorial we will dig deeper into property values (VALP) using these operations and ggplot2 charts. We want to explore what factors influence the variables in our dataset. In this third tutorial we have introduced most of the tools we need in order to perform a exploratory data analysis using Spark and R on a large dataset.

And finally, remember that all the code for these series of Spark and R tutorials can be found in its own GitHub repository. Go there and make it yours.