Datasets In Apache Spark - Part 3 | Writing Datasets to Disk

In the last tutorial we've seen how to create parametrized datasets. Once you create datasets and perform some operations on them, you would like to save those results back into storage. This is what we'll try to do in this article - Saving Datasets to storage.

The first thing we'll do as always is to create the spark-session variable.

// Initialize Sparksession
SparkSession spark = SparkSession.builder().appName("Freblogg-Spark").master("local").getOrCreate();

Using that session variable, we read the fake-people.csv file which has data like this:

id,first_name,last_name,email,gender,ip_address
1,Netti,McKirdy,nmckirdy0@slideshare.net,Female,148.3.248.193
2,Nickey,Curreen,ncurreen1@tripadvisor.com,Male,206.9.48.216
3,Allayne,Chatainier,achatainier2@trellian.com,Male,191.118.4.217
...

We read this file into a dataset as following:

// Read csv file
Dataset<Row> peopleDs = spark.read().option("header", "true").csv("fake-people.csv");

After we have the dataset, Let's assume you've performed some operations on it. Some column selections, some filtering, some sorting etc. And we have a new dataset after all those operations.

// After performing several awesome operations
Dataset<Row> newDs = ....

We want to store this dataset back on the disk. We can do that with the write() on spark session variable, just like read().

newDs.write().csv("processed-data");

The processed-data in the above command is not the name for the output CSV file but instead for the output directory. When you write a Dataset to a file, it will store the data in the format you asked for, CSV in this case, along with adding some check files and status flags as well creating a directory with that name.

These are the files that get created in the processed-data folder.

$ ls ../../apache-spark/processed-data
_SUCCESS  part-00000-311049cf-3e48-4286-b93c-7d2096a18678-c000.csv

There are two more hidden CRC files that I'm not showing here. The part-00000-31hxxxxxxxxx.csv is the actual data file which has the data from the new dataset.

You can also create a json file by running

newDs.write().json("processed-data")

And that will create another folder with json file and the _SUCCESS file inside it.

You can also save this data to an external Database if you want to. You'll use the jdbc() method along with the connection string and the table name. And Spark will write it to the DB.

Apart from the CSV and JSON formats, there is one more popular data format in the Data Science and Big Data world. That is Parquet. Parquet is a data format that is highly optimized and well suited for column-wise operations. It is widely used in a lot of projects in the Big Data ecosystem as a data serialization format. And In Spark, Parquet is the default file storage format. Of course one main difference between Parquet and formats like CSV, JSON is that Parquet is not meant to be used for humans. It can only be read by a parquet reader. A sample file looks something like this:

PAR1   �k �>, �          999  1     �5,   �      1   2   3   4   5   6   7   8   9  - 0   1   2   3   4   5   6   7   8   < 2 < 2 < 2 < 2 < 2 < 2 < 2 <
.....

Utterly gibberish. But spark can read and understand it. In fact, As Parquet is designed for speed and throughput, it can be 10-100 times faster than reading/writing from an ordinary data format like CSV or JSON, depending on the type of data.

You save dataset to Parquet as follows:

newDs.write().parquet("processed");

And this will save the dataset as a parquet file along with the _SUCCESS status file.

That is all for this article.

---

For more programming articles, checkout Freblogg, Freblogg/Java, Freblogg/Spark

Articles on Apache Spark:

Map Vs Flat map

Spark Word count with Java

Datasets in Spark | Part I

Datasets in Spark | Part II

---

This is the 17th article as part of my twitter challenge #30DaysOfBlogging. Thirteen more articles on various topics, including but not limited to, Java, Git, Vim, Software Development, Python, to come.

If you are interested in this, make sure to follow me on Twitter @durgaswaroop. While you're at it, Go ahead and subscribe here on medium and my other blog as well.

---

If you are interested in contributing to any open source projects and haven't found the right project or if you were unsure on how to begin, I would like to suggest my own project, Delorean which is a Distributed Version control system, built from scratch in scala. You can contribute not only in the form of code, but also with usage documentation and also by identifying any bugs in its functionality.

---

Thanks for reading. See you again in the next article.

Read More

Datasets In Apache Spark | Part 2

In the two last tutorials we have covered what Apache Spark is and also got ourselves familiar with Datasets in Apache Spark, which is the primary data abstraction in Spark. In this tutorial we will see how to read a data file as a parametrized Bean object Dataset using Encoders.

 

This tutorial is going to be short, but this is very important as you would find yourself doing this frequently. In the last article you've seen how to read a CSV or JSON file as a Dataset. You might have noticed that we were using Dataset<Row> for everything. If you're not familiar with Generics in Java, Dataset<Row> can be thought of as a Dataset consisting of Row objects. The Row object is a spark sql class and is the default when creating a Dataset.

Although the Row class has some useful methods, as a generic object suitable for all types, it is not suitable for everything. Since Datasets usually store data that usually corresponds to a Bean class, it is better to create a Dataset of that bean class instead of Row. With this, you'll have access to all your usual getters and setters of the bean class. That's what We'll do in this article. We'll create a Dataset of POJO's instead of Row objects.

I'm using the same fake-people.csv file that I used in the last article that looks like this:

id,first_name,last_name,email,gender,ip_address
1,Netti,McKirdy,nmckirdy0@slideshare.net,Female,148.3.248.193
2,Nickey,Curreen,ncurreen1@tripadvisor.com,Male,206.9.48.216
3,Allayne,Chatainier,achatainier2@trellian.com,Male,191.118.4.217
...

To represent this data, I've created a POJO called FakePeople.java, which looks like this:

import lombok.Data;
public @Data class FakePeople {
    final int id;
    final private String firstName;
    final private String lastName;
    final private String email;
    final private String gender;
    final private String ipAddress;
}

I'm using Project Lombok here, to generate the required Getters, Setters and other POJO methods. (If you don't know about Lombok, you should definitely check that out. It is quite handy).

We have our POJO now, Let's get a parametrized Dataset. To achieve this we first need to create an Encoder. We do that for the FakePeople class as following:

Encoder<FakePeople> fakePeopleEncoder = Encoders.bean(FakePeople.class);

This will register our encoder which will help us parse our CSV data.

Of course we need our spark session variable as well.

// Initialize Sparksession
SparkSession spark = SparkSession.builder().appName("Freblogg-Spark").master("local").getOrCreate();

Now we can go ahead and read the CSV file, very much like the way we did before with just one addition.

// Without Encoder
Dataset<Row> people = spark.read().option("header", "true").csv("fake-people.csv");

// With Encoder
Dataset<FakePeople> people = spark.read().option("header", "true").csv("fake-people.csv").as(fakePeopleEncoder);

And the output of people.show(5) is the same as what you'd expect.

+---+----------+----------+--------------------+------+--------------+
| id|first_name| last_name|               email|gender|    ip_address|
+---+----------+----------+--------------------+------+--------------+
|  1|     Netti|   McKirdy|nmckirdy0@slidesh...|Female| 148.3.248.193|
|  2|    Nickey|   Curreen|ncurreen1@tripadv...|  Male|  206.9.48.216|
|  3|   Allayne|Chatainier|achatainier2@trel...|  Male| 191.118.4.217|
|  4|     Tades|    Emmett|temmett3@barnesan...|  Male|153.113.87.195|
|  5|     Shawn|    McGenn|smcgenn4@shop-pro.jp|  Male|  247.45.80.68|
+---+----------+----------+--------------------+------+--------------+

As you can see the only difference in creating the Dataset is .as(fakePeopleEncoder) and that gets us Dataset<FakePeople> instead of Dataset<Row>. And with that, we now have access to all the getters, setters of FakePeople class which we wouldn't otherwise have with a Row object. We'll explore more about how this is useful in a future tutorial.

For more information on Datasets: Spark SQL, DataFrames and Datasets Guide

That is all for this article.

---

For more programming articles, checkout Freblogg, Freblogg/Java, Freblogg/Spark

Apache Spark articles:

Word count with Apache Spark and Java

Datasets in Apache Spark | Part 1

Datasets in Apache Spark | Part 2

---

This is the 11th article as part of my twitter challenge #31DaysOfBlogging. Nineteen more articles on various topics, including but not limited to, Java, Git, Vim, Software Development, Python, to come.

If you are interested in this, make sure to follow me on Twitter @durgaswaroop. While you're at it, Go ahead and subscribe here on medium and my other blog as well.

---

If you are interested in contributing to any open source projects and haven't found the right project or if you were unsure on how to begin, I would like to suggest my own project, Delorean which is a Distributed Version control system, built from scratch in scala. You can contribute not only in the form of code, but also with usage documentation and also by identifying any bugs in its functionality.

---

Thanks for reading. See you again in the next article.

Read More

Datasets in Apache Spark | Part 1

In my previous post I have talked about Apache Spark. We have also built an application for counting the number of words in a file, which is the hello world equivalent of the big data world.

It has been over 18 months since that article and spark has changed quite a lot in this time. A new major release of spark, which is spark-2.0 came out and now the latest version is 2.2.1. And with a new version comes new API’s and improvements. In-fact the first thing you’ll probably notice is that, you don’t need to create SparkContext or JavaSparkContext objects anymore. The various context and configurations have been put together into a new class SparkSession. You can still access the SparkContext or the SqlContext from the SparkSession object itself. So, you’ll be starting your programs with this now:

SparkSession spark = SparkSession.builder().appName("Freblogg-Spark").master("local").getOrCreate();

And you can use this spark variable the way you’d use other context variables.

Another change in Spark 2.0 is that, there is a heavy emphasis on the usage of Dataset API’s, and for a good reason. Datasets are more performant and memory efficient than RDD’s. RDD (Resilient Distributed Datasets) have been pushed to second place now. You can still use RDD’s if you want but Datasets are the preferred API. In fact, datasets have some nice convenience methods that we can use them for even unstructured data like text as well. Let’s generate some cool lipsum from Malevole. It looks something like this:

Ulysses, Ulysses - Soaring through all the galaxies. In search of Earth, flying in to the night. Ulysses, Ulysses - Fighting evil 
and tyranny, with all his power, and with all of his might. Ulysses - no-one else can do the things you do. Ulysses - like a bolt of
thunder from the blue. Ulysses - always fighting all the evil forces bringing peace and justice to all....

Now, you might try to use an RDD to read this, but let’s see what we can do with Datasets.

Dataset<String> lipsumDs = spark.read().textFile("fake-text.txt");
lipsumDs.show(5);

Here we are reading the text file using the spark object we created earlier and that gives us a Dataset<String> lipsumDs. The show() method on the dataset object prints the dataset. And we get the following output:

+--------------------+
|               value|
+--------------------+
|Ulysses, Ulysses ...|
|Ulysses, Ulysses ...|
| no-one else can ...|
|  always fighting...|
|                    |
+--------------------+

What we see here are the lines of the text file. Each line in the file is now a row in the Dataset. There are now a rich set of functions available to you in Datasets which weren’t in RDD’s. You can do filters on the rows for certain words, do a count on the table, perform groupBy operations, etc all like you would on a Database table. For a full list of all the available operations on Dataset, read this: Dataset: Spark Documentation.

I hope that’s enough talk about unstructured data analysis. Let’s get to the main focus of this article, which is using Datasets for structured data. More specifically, csv and json. For this tutorial, I am using the data created from Mockaroo, an online data generator. I’ve created 1000 csv records that look like this:

id,first_name,last_name,email,gender,ip_address
1,Netti,McKirdy,nmckirdy0@slideshare.net,Female,148.3.248.193
2,Nickey,Curreen,ncurreen1@tripadvisor.com,Male,206.9.48.216
3,Allayne,Chatainier,achatainier2@trellian.com,Male,191.118.4.217
4,Tades,Emmett,temmett3@barnesandnoble.com,Male,153.113.87.195
5,Shawn,McGenn,smcgenn4@shop-pro.jp,Male,247.45.80.68
6,Giuseppe,Scobbie,gscobbie5@twitter.com,Male,123.114.131.200
...

We’ll use this data, which I’ve put in a file named fake-people.csv, to work with Datasets. Let’s create a Dataset out of this csv data.

Dataset<Row> peopleDs = spark.read().option("header", "true").csv("fake-people.csv");
peopleDs.show(5);

Since we’ve column headers in our data, we add the .option("header", "true") and the output is a nicely formatted table of the data with all the columns like this:

+---+----------+----------+--------------------+------+--------------+
| id|first_name| last_name|               email|gender|    ip_address|
+---+----------+----------+--------------------+------+--------------+
|  1|     Netti|   McKirdy|nmckirdy0@slidesh...|Female| 148.3.248.193|
|  2|    Nickey|   Curreen|ncurreen1@tripadv...|  Male|  206.9.48.216|
|  3|   Allayne|Chatainier|achatainier2@trel...|  Male| 191.118.4.217|
|  4|     Tades|    Emmett|temmett3@barnesan...|  Male|153.113.87.195|
|  5|     Shawn|    McGenn|smcgenn4@shop-pro.jp|  Male|  247.45.80.68|
+---+----------+----------+--------------------+------+--------------+

You can read in json data similarly as well. So, I generated some json this time from mockaroo.

{"id":1,"first_name":"Zenia","last_name":"Joberne","email":"zjoberne0@foxnews.com","gender":"Female","ip_address":"214.207.159.43"}
{"id":2,"first_name":"Renard","last_name":"Kezor","email":"rkezor1@elpais.com","gender":"Male","ip_address":"199.3.18.104"}
{"id":3,"first_name":"Briant","last_name":"Patel","email":"bpatel2@odnoklassniki.ru","gender":"Male","ip_address":"111.184.217.23"}
{"id":4,"first_name":"Robinett","last_name":"Heasley","email":"rheasley3@tiny.cc","gender":"Female","ip_address":"21.40.190.226"}
{"id":5,"first_name":"Rosalinda","last_name":"Glandfield","email":"rglandfield4@indiegogo.com","gender":"Female","ip_address":"26.16.4.132"}
{"id":6,"first_name":"Haslett","last_name":"Culligan","email":"hculligan5@meetup.com","gender":"Male","ip_address":"201.191.72.10"}
....

Note: Spark can read json only of this format where we have one object per row. Otherwise you will see _corrupt_record when you print your dataset. That’s your cue to make sure the json is formatted as per spark’s need.

And you read json very similar to the way you read csv. Since in json we don’t have headers, we don’t need the header option.

Dataset<Row> peopleJsonDs = spark.read().json("fake-people.json");
peopleJsonDs.show(5);

And the output is,

+--------------------+----------+------+---+--------------+---------+
|               email|first_name|gender| id|    ip_address|last_name|
+--------------------+----------+------+---+--------------+---------+
|psurgison0@istock...|   Prissie|Female|  1| 48.151.89.171| Surgison|
| rsewell1@jalbum.net|    Robena|Female|  2| 184.16.37.210|   Sewell|
|aluxon2@list-mana...| Annamarie|Female|  3| 254.69.187.23|    Luxon|
|sodoherty3@twitpi...|   Shannah|Female|  4| 0.245.101.197|O'Doherty|
| alodford4@jigsy.com|     Alice|Female|  5|70.217.170.182|  Lodford|
+--------------------+----------+------+---+--------------+---------+

You can see the order of columns is jumbled. This is because JSON data doesn’t usually keep any specified order and so, when you read JSON data into a dataset, the order might not be same as what you’ve given. Of course if you want to display the columns in a particular order, you can always do a select operation.

peopleJsonDs.select("id", "first_name", "last_name", "email", "gender", "ip_address").show(5);

And that would print it in the right order. This is exactly like the SELECT query in SQL, if you’re familiar with it.

Now, that we have seen how to create Datasets, let’s see some of the operations we can perform on them.

Operations on Datasets

Datasets are built on top of Dataframes. So, if you’re already familiar with Dataframes in the spark 1.x releases you already know a ton about Datasets. Some of the operations you can perform on Dataset are as follows:

Column selection

Select one or more columns from the dataset.

peopleDs.select("email").show(5); // Selecting one column
peopleDs.select(col("email"), col("gender")).show(5); // Selecting multiple columns

Note: col is a static import of org.apache.spark.sql.functions.col;

Filtering on columns

Filter a subset of rows in the dataset based on conditions.

// Filter rows with id > 5 and \<= 10
peopleDs.filter(col("id").$less$eq(10).and(col("id").$greater(5))).show();

Dropping columns

Remove one or more columns from the dataset

peopleDs.drop("last_name", "ip_address").show(5);

Sorting on columns

peopleDs.sort(desc("first_name")).show(5);

And that sorts the dataset in the reverse order of the column first_name.

Output:

+---+----------+---------+--------------------+------+-------------+
| id|first_name|last_name|               email|gender|   ip_address|
+---+----------+---------+--------------------+------+-------------+
|685|  Zedekiah|  Brockie|zbrockiej0@mozill...|  Male|105.119.18.98|
|308|     Zarla| Bryceson|zbryceson8j@redif...|Female|55.118.168.15|
|636|  Zacherie|   Kermon|zkermonhn@prnewsw...|  Male| 120.36.10.87|

Those are some of the functions that you can use with Datasets. There are still several Database table type operations on Datasets, like group By, aggregations, joins, etc. We’ll look at them in the next article on Spark as I think this article already has a lot of information already and I don’t want to overload you with information.

So, that is all for this article. If you’re someone that has never tried Datasets or Dataframes, I hope this article gave a good introduction on the topic to keep you interested in learning more.

The full code is available as gist.

---

For more Java, Apache Spark, Big data and other programming articles, checkout Freblogg, Freblogg/Java, Freblogg/Spark

---

This is the fifth article as part of my twitter challenge #30DaysOfBlogging. Twenty-five more articles on various topics including but not limited to Java, Git, Vim, Software Development, Python, to come.

If you are interested in this, make sure to follow me on Twitter @durgaswaroop. While you’re at it, Go ahead and subscribe to this blog and my blog on Medium as well.

---

If you are interested in contributing to any open source projects and haven’t found the right project or if you were unsure on how to begin, I would like to suggest my own project, Delorean which is a Distributed Version control system, built from scratch in scala. You can contribute not only in the form of code, but also with usage documentation and also by identifying any bugs in the functionality.

---

Thanks for reading. See you again in the next article.

Read More

Word Count application with Apache Spark and Java

Apache Spark is becoming ubiquitous by day and has been dubbed the next big thing in the Big Data world. Spark has been replacing MapReduce with its speed and scalability. In this Spark series we will try to solve various problems using Spark and Java.

Word count program is the big data equivalent of the classic Hello world program. The aim of this program is to scan a text file and display the number of times a word has occurred in that particular file. And for this word count application we will be using Apache spark 1.6 with Java 8.

Read More

Apache Spark | Map and FlatMap

Map and FlatMap functions transform one collection in to another just like the map and flatmap functions in several other functional languages. In the context of Apache Spark, they transform one RDD in to another RDD.

Here is how they differ from each other.

Read More