Introduction

In this lesson, we will explore how to create Datasets in Apache Spark using Scala. Datasets are a strong type of distributed collection, providing the benefits of both RDDs (Resilient Distributed Datasets) and DataFrames, with the added benefit of compile-time type safety. They are designed to work with structured data and can be constructed from a variety of data sources.

We will cover different methods for creating Datasets, including how to create them from existing collections using Scala case classes and Java beans, how to convert DataFrames to Datasets (and vice versa), the concept of defining encoders, and how to read structured data formats like JSON and Parquet into Datasets. We will illustrate each of these concepts with practical examples.

Creating Datasets from Collections

To create a Dataset in Spark, we often start with an existing collection, such as a list or array. The following example demonstrates how to create a Dataset from a collection of Scala case classes.

Example: Creating a Dataset from Scala Case Classes

First, let's define a simple case class to represent a person.

case class Person(name: String, age: Int)

Now, we can create a collection of Person objects and convert it into a Dataset:

import org.apache.spark.sql.SparkSession
import spark.implicits._

val spark = SparkSession.builder()
  .appName("Creating Datasets")
  .master("local[*]")
  .getOrCreate()

val peopleList = List(Person("Alice", 25), Person("Bob", 29), Person("Charlie", 23))
val peopleDS = spark.createDataset(peopleList)

Explanation

In this example, we start by importing necessary libraries and creating a SparkSession. We define a list of Person instances and call spark.createDataset() to convert the Scala List into a Dataset. The type of the Dataset is inferred from the case class.

Output

To display the Dataset, we can use:

peopleDS.show()

The output will be:

+-------+---+
|   name|age|
+-------+---+
|  Alice| 25|
|    Bob| 29|
|Charlie| 23|
+-------+---+

Converting DataFrame to Dataset and Vice Versa

Datasets provide enhanced type safety compared to DataFrames, and sometimes you might need to convert between these two formats. To convert a DataFrame to a Dataset, you can use the .as[] method.

Example: Converting DataFrame to Dataset

Suppose we already have a DataFrame:

val df = peopleDS.toDF()

To convert it back into a Dataset, we would utilize:

val peopleDSFromDF = df.as[Person]

Explanation

Here we use the as[Type] method to cast the DataFrame back into a Dataset of the specified type, Person. This provides the benefits of type safety and compile-time checks.

Defining Encoders

Encoders are a critical part of working with Datasets in Spark. They enable Spark to serialize the objects of a specific type, facilitating their processing. When using Scala case classes, Spark automatically derives Encoders for us.

Example: Using Custom Encoder

If you’re using Java beans or other types, you might need to define an encoder manually.

import org.apache.spark.sql.Encoders

val customEncoder = Encoders.bean(classOf[Person])
val peopleDSWithEncoder = spark.createDataset(peopleList)(customEncoder)

Explanation

In this example, we explicitly define an encoder for our Java Bean (or similar structured data type). This allows Spark to understand how to serialize these objects properly.

Reading Structured Data into Datasets

Spark allows reading structured data from various formats directly into a Dataset. We will see how to read JSON and Parquet files into Datasets.

Example: Reading JSON Data

Assuming we have a JSON file, people.json, with the following content:

[
    {"name": "Alice", "age": 25},
    {"name": "Bob", "age": 29},
    {"name": "Charlie", "age": 23}
]

We can read this JSON file into a Dataset as follows:

val jsonDS = spark.read.json("path/to/people.json").as[Person]
jsonDS.show()

Output

The output will match the structure of the JSON file, presenting:

+-------+---+
|   name|age|
+-------+---+
|  Alice| 25|
|    Bob| 29|
|Charlie| 23|
+-------+---+

Reading Parquet Data

Reading a Parquet file is similar. For a Parquet file containing similar structure, you could do:

val parquetDS = spark.read.parquet("path/to/people.parquet").as[Person]
parquetDS.show()

The output will be structured similarly to previous examples, leveraging the schema stored in the Parquet file.

Summary

In this lesson, we learned how to create Datasets in Spark using Scala. We covered multiple methods including creation from collections, conversion between DataFrames and Datasets, defining custom Encoders, and reading structured data from JSON and Parquet formats into Datasets. Understanding Datasets is crucial for manipulating structured data efficiently and safely in Apache Spark.