Apache Spark is a fast, in-memory data processing engine with elegant and expressive development APIs that enable data workers to efficiently execute streaming, machine learning or SQL workloads. Spark Structured APIs are a toolset for handling structured data in Spark. They are built on top of the Spark core, and they provide a programming interface for working with structured and semi-structured data. The Structured APIs include the Dataset, DataFrame, and SQL APIs, which are available in Spark's SQL module. Below is an introduction to these Structured APIs:
- **DataFrame API:**
- A DataFrame is a distributed collection of data organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R or Python, but with optimizations under the hood for support for big data. It is the most common Structured API.
- You can think of a DataFrame as a spreadsheet with named columns.
- DataFrames are immutable — once created, they cannot be changed. To transform a DataFrame, you create a new one using transformations.
- They allow you to run SQL-like queries, and they also support powerful, expressive operations for filtering, grouping, and aggregating data.
- **Dataset API:**
- A Dataset is a distributed collection of data that provides the benefits of strong typing, ability to use powerful lambda functions, with the benefits of Spark’s DataFrame API.
- It is a type-safe, object-oriented programming interface.
- Datasets are a strictly more powerful abstraction over DataFrames because they have all the benefits of DataFrames and also allow you to use functional transformations on a distributed collection of strongly typed objects.
- They are used to achieve high-level abstractions and optimizations, while also maintaining type-safety.
- **Spark SQL:**
- Spark SQL is a Spark module for structured data processing that provides a programming interface for data querying.
- You can run SQL queries on data that is stored in various formats (like Parquet, Avro, JSON, and Hive), and the results are returned as a DataFrame.
- With Spark SQL, you can also mix SQL queries with Spark programs, thus combining the benefits of both declarative SQL and imperative APIs (like the DataFrame and Dataset APIs).
- Spark SQL also includes a powerful Catalyst optimizer that helps to optimize queries to execute them more efficiently.
- **Streaming Queries:**
- With Structured Streaming, you can work with real-time data within the same APIs that you use for batch processing. It’s a scalable and fault-tolerant stream processing engine built on the Spark SQL engine.
- Your queries on streaming dataframes/datasets will be run incrementally and continuously updated as real-time data flows in.
- **Catalyst Optimizer:**
- Under the hood of these APIs, Spark SQL uses a very sophisticated optimization engine called Catalyst. It uses rule-based transformations to optimize the logical plan of the computation and generates an efficient physical plan for the execution on the cluster.
- **Integration with Various Data Sources:**
- Structured APIs can efficiently read and write data from a variety of structured sources, including Hive, Avro, Parquet, ORC, JSON, and JDBC.
These Structured APIs are designed to make it easy to build end-to-end big data processing pipelines that combine batch, streaming, and interactive queries. They provide a high-level, expressive API for data analysis and enable efficient execution across distributed data sources.
Introduction to DataFrame API
A DataFrame in Spark is a distributed collection of data organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R or Python, but with optimizations under the hood for support with big data. DataFrames can be constructed from a wide variety of sources such as structured data files, Hive tables, external databases, or existing RDDs (Resilient Distributed Datasets). Here are some key features of the DataFrame API:
- **Immutability:**
- Once created, a DataFrame can’t be changed. Instead, any transformation on a DataFrame returns a new DataFrame.
- **Lazy Evaluation:**
- Transformations on DataFrames are not computed immediately. Spark waits until it sees an action (like a `collect` or a `count` operation) and then it computes the result. This approach, called lazy evaluation, enables Spark to optimize the execution plan more effectively.
- **Optimized Execution Plans:**
- DataFrames are built on top of Spark SQL’s Catalyst optimizer, which optimizes queries with a logical and physical execution plan. This often results in much faster execution compared to RDDs.
- **Interoperability:**
- You can run SQL queries on DataFrames, convert them to Datasets (if you are using a statically typed language like Scala or Java), and you can seamlessly mix imperative (DataFrame / Dataset) and declarative (SQL) APIs.
- **Various Data Sources:**
- DataFrames can be created from various formats (Parquet, Avro, JSON, and Hive), and various data sources (like HDFS, Apache Cassandra, Apache Kafka, and many others).
Let's assume we have a file named `people.json` with the following content:
- heading
- ### Examples with DataFrame API
- code
-
[ {"name": "John", "age": 30, "city": "New York"}, {"name": "Anna", "age": 29, "city": "London"}, {"name": "Mike", "age": 32, "city": "San Francisco"} ]
Here are some examples that show how to use the DataFrame API:
- heading
- **Creating a DataFrame from a JSON file:**
- code
-
from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("DataFrame Example") \ .getOrCreate() # Load a JSON file people = spark.read.json("people.json")
- heading
- Showing the content of the DataFrame
- code
-
# Show the content of the DataFrame people.show()
- heading
- This will output:
- code
-
+----+-------+-------------+ | age| city| name| +----+-------+-------------+ | 30|New York| John| | 29| London| Anna| | 32|San Francisco| Mike| +----+-------+-------------+
- heading
- Running SQL-like queries
- code
-
# Register the DataFrame as a SQL temporary view people.createOrReplaceTempView("people") # Use SQL to query the data result = spark.sql("SELECT * FROM people WHERE age >= 30") # Show the result of the SQL query result.show()
- heading
- This will output:
- code
-
+----+-------------+----+ | age| city|name| +----+-------------+----+ | 30| New York|John| | 32|San Francisco|Mike| +----+-------------+----+
- heading
- Applying some transformations
- code
-
# Use DataFrame API to filter rows older_than_30 = people.filter(people.age > 30) # Show the filtered rows older_than_30.show()
- heading
- This will output:
- code
-
+---+-------------+----+ |age| city|name| +---+-------------+----+ | 32|San Francisco|Mike| +---+-------------+----+
These examples are in PySpark (Spark Python API), but the DataFrame API is also available in Scala, Java, and R, with similar functions and syntax.
Please note that in a real Spark application, you'll need to setup and configure a `SparkSession`, and the data files need to be accessible from the Spark cluster. The method to read data (in this example, `read.json`) would change depending on the data format you are using.
- heading
- how to create a DataFrame from a list of data using PySpark,
- code
-
from pyspark.sql import SparkSession from pyspark.sql import Row spark = SparkSession.builder \ .appName("Create DataFrame from a list") \ .getOrCreate() data = [("John", 30, "New York"), ("Anna", 29, "London"), ("Mike", 32, "San Francisco")] columns = ["Name", "Age", "City"] df = spark.createDataFrame(data, columns) rows = [Row(Name=name, Age=age, City=city) for name, age, city in data] df = spark.createDataFrame(rows) rows = [Row(Name=name, Age=age, City=city) for name, age, city in data] df = spark.createDataFrame(rows) df.show()
- heading
- result is
- code
-
+----+---+-------------+ |Name|Age| City| +----+---+-------------+ |John| 30| New York| |Anna| 29| London| |Mike| 32|San Francisco| +----+---+-------------+
- heading
- remember to stop spark
- code
-
spark.stop()
Grouping data in Spark is a very common operation that allows you to aggregate data by specific columns, similar to the "GROUP BY" clause in SQL.
Once you group the data, you can run various aggregate functions on each group of data, such as counting the number of elements in each group, finding the average value, summing the values, finding the maximum or minimum values, and more.
Below is a step-by-step introduction to grouping data in PySpark using the DataFrame API:
- heading
- **Step 1:** Import the necessary modules and create a `SparkSession`:
- code
-
from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("Grouping Data Example") \ .getOrCreate()
- heading
- **Step 2:** Create a DataFrame.
- code
-
data = [("John", 30, "New York"), ("Anna", 29, "London"), ("Mike", 32, "San Francisco"), ("Mike", 27, "San Jose"), ("John", 40, "New York"), ("Anna", 41, "London")] columns = ["Name", "Age", "City"] df = spark.createDataFrame(data, columns)
- heading
- **Step 3:** Group the data. In this example, we are grouping the data by the "City" column:
- code
-
grouped_data = df.groupBy("City")
- heading
- Apply aggregate functions to the grouped data. Here are some examples:
- code
-
result1 = grouped_data.count() result1.show()
- heading
- This will output:
- code
-
``` +-------------+-----+ | City|count| +-------------+-----+ | San Jose| 1| | New York| 2| |San Francisco| 1| | London| 2| +-------------+-----+ ```
- heading
- - **Calculate the average age of people in each city:**
- code
-
result2 = grouped_data.avg("Age") result2.show()
- heading
- This will output:
- code
-
+-------------+--------+ | City|avg(Age)| +-------------+--------+ | San Jose| 27.0| | New York| 35.0| |San Francisco| 32.0| | London| 35.0| +-------------+--------+
- heading
- - **Find the maximum and minimum age of people in each city:**
- code
-
result3 = grouped_data.agg({"Age": "max", "Age": "min"}) result3.show()
- heading
- This will output:
- code
-
+-------------+--------+ | City|min(Age)| +-------------+--------+ | San Jose| 27| | New York| 30| |San Francisco| 32| | London| 29| +-------------+--------+
In these examples:
- `groupBy("City")` groups the data by the "City" column, which creates a `GroupedData` object.
- Various aggregate functions like `count()`, `avg("Age")`, and `agg({"Age": "max", "Age": "min"})` are applied to the `GroupedData` object.
- `show()` is used to display the result of these aggregate functions.
This gives you a sense of some of the powerful operations you can perform once you have grouped your data using PySpark's DataFrame API.