We are going to look at five join types available in dplyr: innerjoin, semijoin, leftjoin, antijoin and fulljoin. We are going to examine the output of each join type using a simple example. In the fifth section we’ll learn how to combine the dplyr and ggplot2 (using chaining) commands to build expressive charts and graphs.
dplyr isan R package for working with structured data both in and outside of R.dplyr makes data manipulation for R users easy, consistent, andperformant. With dplyr as an interface to manipulating Spark DataFrames,you can:
Statements in dplyr can be chained together using pipes defined by themagrittrR package. dplyr also supports non-standardevalutionof its arguments. For more information on dplyr, see theintroduction,a guide for connecting todatabases,and a variety ofvignettes.
You can read data into Spark DataFrames using the followingfunctions:
|Reads a CSV file and provides a data source compatible with dplyr|
|Reads a JSON file and provides a data source compatible with dplyr|
|Reads a parquet file and provides a data source compatible with dplyr|
Regardless of the format of your data, Spark supports reading data froma variety of different data sources. These include data stored on HDFS(
hdfs:// protocol), Amazon S3 (
s3n:// protocol), or local filesavailable to the Spark worker nodes (
Each of these functions returns a reference to a Spark DataFrame whichcan be used as a dplyr table (
This guide will demonstrate some of the basic data manipulation verbs ofdplyr by using data from the
nycflights13 R package. This packagecontains data for all 336,776 flights departing New York City in 2013.It also includes useful metadata on airlines, airports, weather, andplanes. The data comes from the US Bureau of TransportationStatistics,and is documented in
Connect to the cluster and copy the flights data using the
copy_tofunction. Caveat: The flight data in
nycflights13 is convenient fordplyr demonstrations because it is small, but in practice large datashould rarely be copied directly from R objects.
Verbs are dplyr commands for manipulating data. When connected to aSpark DataFrame, dplyr translates the commands into Spark SQLstatements. Remote data sources use exactly the same five verbs as localdata sources. Here are the five verbs with their corresponding SQLcommands:
aggregators: sum, min, sd, etc.
operators: +, *, log, etc.
When working with databases, dplyr tries to be as lazy as possible:
It never pulls data into R unless you explicitly ask for it.
It delays doing any work until the last possible moment: it collectstogether everything you want to do and then sends it to the databasein one step.
For example, take the followingcode:
This sequence of operations never actually touches the database. It’snot until you ask for the data (e.g. by printing
c4) that dplyrrequests the results from the database.
You can usemagrittrpipes to write cleaner syntax. Using the same example from above, youcan write a much cleaner version like this:
group_by function corresponds to the
GROUP BY statement in SQL.
You can copy data from Spark into R’s memory by using
collect() executes the Spark query and returns the results to R forfurther analysis and visualization.
It’s relatively straightforward to translate R code to SQL (or indeed toany programming language) when doing simple mathematical operations ofthe form you normally use when filtering, mutating and summarizing.dplyr knows how to convert the following R functions to Spark SQL:
dplyr supports Spark SQL window functions. Window functions are used inconjunction with mutate and filter to solve a wide range of problems.You can compare the dplyr syntax to the query it has generated by using
It’s rare that a data analysis involves only a single table of data. Inpractice, you’ll normally have many tables that contribute to ananalysis, and you need flexible tools to combine them. In dplyr, thereare three families of verbs that work with two tables at a time:
Mutating joins, which add new variables to one table from matchingrows in another.
Filtering joins, which filter observations from one table based onwhether or not they match an observation in the other table.
Set operations, which combine the observations in the data sets asif they were set elements.
All two-table verbs work similarly. The first two arguments are
y, and provide the tables to combine. The output is always a new tablewith the same type as
The following statements are equivalent:
You can use
sample_frac() to take a random sample ofrows: use
sample_n() for a fixed number and
sample_frac() for afixed fraction.
It is often useful to save the results of your analysis or the tablesthat you have generated on your Spark cluster into persistent storage.The best option in many scenarios is to write the table out to aParquet file using thespark_write_parquetfunction. For example:
This will write the Spark DataFrame referenced by the tbl R variable tothe given HDFS path. You can use thespark_read_parquetfunction to read the same table back into a subsequent Sparksession:
You can also write data as CSV or JSON using thespark_write_csv andspark_write_jsonfunctions.
Many of Hive’s built-in functions (UDF) and built-in aggregate functions(UDAF) can be called inside dplyr’s mutate and summarize. The LanguangeReferenceUDFpage provides the list of available functions.
The following example uses the datediff and current_date HiveUDFs to figure the difference between the flight_date and the currentsystem date:
These are methods for the dplyr join generics. They are translatedto the following SQL queries:
SELECT * FROM x JOIN y ON x.a = y.a
SELECT * FROM x LEFT JOIN y ON x.a = y.a
SELECT * FROM x RIGHT JOIN y ON x.a = y.a
SELECT * FROM x FULL JOIN y ON x.a = y.a
SELECT * FROM x WHERE EXISTS (SELECT 1 FROM y WHERE x.a = y.a)
SELECT * FROM x WHERE NOT EXISTS (SELECT 1 FROM y WHERE x.a = y.a)
A pair of lazy data frames backed by database queries.
A character vector of variables to join by.
To join by different variables on
To join by multiple variables, use a vector with length > 1.For example,
To perform a cross-join, generating all combinations of
This allows you to join tables across srcs, but it's potentially expensiveoperation so you must opt into it.
If there are non-joined duplicate variables in
Other parameters passed onto methods.
A custom join predicate as an SQL expression.Usually joins use column equality, but you can perform more complexqueries by supply
Should NA (NULL) values match one another?The default, 'never', is how databases usually work.
show_query() to see the generatedquery, and use
collect() to execute the queryand return data to R.