Unit 4: Running SQL Queries Using Spark SQL - Subjective Questions

Spark SQL is a module of Apache Spark used for processing structured and semi-structured data. It allows users to query data using SQL as well as Spark programming APIs.

Importance of Spark SQL:

• It provides a familiar SQL interface for querying large datasets.
• It supports structured data processing with schema information.
• It integrates SQL queries with Spark programs written in Scala, Java, Python, and R.
• It can process data from sources such as Hive, JSON, Parquet, and JDBC.
• It uses Spark's distributed execution engine, making query processing scalable and fault tolerant.
• It enables optimization through the Catalyst optimizer and Tungsten execution engine.

The major features of Spark SQL include:

• SQL interface: Users can write standard SQL queries to analyze distributed data.
• DataFrame and Dataset APIs: Structured data can be represented using DataFrames and Datasets.
• Schema support: Spark SQL understands the names and data types of columns.
• Multiple data sources: It supports JSON, CSV, Parquet, ORC, Hive tables, and relational databases through JDBC.
• Query optimization: The Catalyst optimizer improves query execution plans automatically.
• Integration with Spark: SQL queries can be combined with RDD, DataFrame, and Dataset operations.
• Fault tolerance: Queries run on Spark's distributed and fault-tolerant execution engine.
• In-memory processing: Frequently used data can be cached to improve performance.

Spark SQL processes a query through several stages:

• Query submission: The user submits an SQL query or uses the DataFrame or Dataset API.
• Parsing: Spark parses the SQL statement and creates an unresolved logical plan.
• Analysis: The analyzer checks tables, columns, functions, and data types using catalog information.
• Optimization: The Catalyst optimizer generates an efficient logical and physical execution plan.
• Code generation: The Tungsten execution engine generates optimized code and manages memory efficiently.
• Execution: The physical plan is divided into stages and tasks, which are distributed across the cluster.
• Result generation: Executors process the data and return the result to the driver or write it to storage.

This process allows Spark SQL to combine declarative SQL processing with distributed execution.

A DataFrame is a distributed collection of data organized into named columns, similar to a table in a relational database.

Differences between RDDs and DataFrames:

• Structure: An RDD is an unstructured collection of objects, whereas a DataFrame has rows and named columns.
• Schema: RDDs do not require a schema, while DataFrames contain schema information.
• Optimization: DataFrame operations are optimized by Catalyst and Tungsten; RDD operations generally receive fewer automatic optimizations.
• Ease of use: DataFrames support SQL queries and concise relational operations.
• Type handling: RDDs provide compile-time type information in typed languages, while DataFrame columns are accessed by name.
• Performance: DataFrames are often faster for structured operations because Spark can optimize their execution.

An RDD can be converted into a DataFrame using several methods:

• Using reflection: Convert an RDD of case classes into a DataFrame. Spark automatically infers the schema from the case class.

  Example: spark.createDataFrame(personRDD) where each element of personRDD is a Person object.

• Using an explicit schema: Create an RDD of Row objects, define a StructType schema, and call createDataFrame.

  This method is useful when the schema is not known at compile time or must be controlled explicitly.

• Using tuples: Convert an RDD of tuples to a DataFrame and assign column names with toDF().

  Example: rdd.toDF("name", "age").

• Using the Java or Python APIs: In Java and Python, rows and schemas can be created explicitly using the corresponding Spark SQL types and session methods.

The selected method depends on whether the schema can be inferred or must be specified manually.

In the reflection-based method, Spark uses the structure of a class, commonly a Scala case class, to infer the schema of an RDD.

General procedure:

• Define a case class containing the required fields.
• Create an RDD whose elements are objects of that case class.
• Import the required implicits from the Spark session.
• Convert the RDD using toDF() or createDataFrame().

Advantages:

• The schema is generated automatically.
• The code is concise and easy to read.
• It is convenient when the data structure is known during program compilation.

Limitations:

• It is mainly suitable when a supported class or case class describes the data.
• It is less flexible when the schema is dynamic.
• Incorrect field definitions can lead to an incorrect inferred schema.
• It may not be convenient for data whose structure is determined at runtime.

An explicit schema conversion is used when the structure of the data is not available through reflection.

Steps:

• Create an RDD of Row objects. Each row contains values in the required column order.
• Define a schema using StructType and StructField objects.
• Specify the column names, data types, and nullability.
• Call spark.createDataFrame(rowRDD, schema).

Example structure:

• A row may contain Row("Asha", 21).
• The schema may define name as StringType and age as IntegerType.
• Spark combines the rows and schema to produce a DataFrame.

Benefits:

• The developer has complete control over column names and data types.
• It supports runtime-generated schemas.
• It avoids ambiguity in type inference.

The order and data types of values in each Row must match the declared schema.

A temporary view is a session-scoped table-like representation of a DataFrame. It allows the DataFrame to be queried using SQL syntax.

Procedure:

• Load or create a DataFrame.
• Register it as a temporary view using createOrReplaceTempView().
• Submit an SQL query through spark.sql().
• Store or display the resulting DataFrame.

Example:

orders.createOrReplaceTempView("orders")
val result = spark.sql("SELECT customer, amount FROM orders WHERE amount > 1000")

A temporary view exists only within the current Spark session. A global temporary view can be accessed across sessions using the global temporary database, but it is removed when the application ends.

A join combines rows from two DataFrames or tables based on a related column or condition.

Common join types:

• Inner join: Returns only rows with matching values in both tables.
• Left outer join: Returns all rows from the left table and matching rows from the right table. Missing matches contain null values.
• Right outer join: Returns all rows from the right table and matching rows from the left table.
• Full outer join: Returns all rows from both tables and fills unmatched columns with null values.
• Cross join: Produces the Cartesian product of the two tables.
• Left semi join: Returns rows from the left table that have a match in the right table, without including right-side columns.
• Left anti join: Returns rows from the left table that have no match in the right table.

Joins should use suitable keys and filters to avoid unnecessary data movement.

Assume Students contains student records and Results contains marks, joined using student_id.

• Inner join:
  SELECT * FROM Students s INNER JOIN Results r ON s.student_id = r.student_id
  returns only students who have a matching result.

• Left outer join:
  SELECT * FROM Students s LEFT JOIN Results r ON s.student_id = r.student_id
  returns every student. Students without results have null values for result columns.

• Full outer join:
  SELECT * FROM Students s FULL OUTER JOIN Results r ON s.student_id = r.student_id
  returns every student and every result record. Unmatched values on either side are represented by null.

The choice depends on the required completeness of the result. Inner joins are useful for matching records, while outer joins preserve unmatched records.

The GROUP BY clause divides rows into groups having the same values in one or more columns. Aggregate functions are then applied independently to each group.

Example:

SELECT department, AVG(salary) AS average_salary
FROM employees
GROUP BY department

This query creates one group for each department and calculates the average salary of that department.

Important points:

• Columns in the SELECT list must generally be grouped or used inside an aggregate function.
• Multiple grouping columns can be specified.
• WHERE filters rows before grouping.
• HAVING filters groups after aggregate values are calculated.
• Grouping is distributed across the cluster, so it may cause a shuffle operation.

GROUP BY is useful for summaries such as department-wise totals, product-wise sales, and region-wise counts.

Both WHERE and HAVING filter data, but they operate at different stages.

• WHERE: Filters individual rows before grouping and aggregation.
• HAVING: Filters groups after grouping and aggregation.

Example:

SELECT department, COUNT(*) AS employee_count
FROM employees
WHERE status = 'active'
GROUP BY department
HAVING COUNT(*) > 10

In this query:

• WHERE status = 'active' removes inactive employee rows before grouping.
• GROUP BY department creates department groups.
• HAVING COUNT(*) > 10 retains only departments with more than ten active employees.

Using WHERE whenever possible can reduce the amount of data processed before the aggregation stage.

The ORDER BY clause sorts the complete result set according to one or more columns.

Example:

SELECT name, salary
FROM employees
ORDER BY salary DESC, name ASC

This sorts employees by salary in descending order and uses name as a secondary ascending key.

ORDER BY versus SORT BY:

• ORDER BY produces a globally ordered result across all partitions.
• Global ordering may require substantial data movement and can become a bottleneck for large datasets.
• SORT BY sorts records within each partition, so the complete output may not be globally ordered.
• SORT BY can be faster and is useful when partition-local ordering is sufficient.

ASC specifies ascending order and DESC specifies descending order.

Aggregate functions calculate a single summary value from multiple rows or calculate one value for each group.

Common aggregate functions include:

• COUNT: Counts rows or non-null values. Example: COUNT(*).
• SUM: Calculates the total of a numeric column. Example: SUM(amount).
• AVG: Calculates the arithmetic mean. Example: AVG(salary).
• MIN: Finds the smallest value. Example: MIN(price).
• MAX: Finds the largest value. Example: MAX(price).
• COUNT DISTINCT: Counts unique values. Example: COUNT(DISTINCT customer_id).

Example query:

SELECT product_id, COUNT(*) AS orders, SUM(amount) AS revenue, MAX(amount) AS largest_order
FROM orders
GROUP BY product_id

Aggregate functions are commonly combined with GROUP BY and HAVING.

Assume an orders table has the columns product_id and amount. The required query is:

SELECT
    product_id,
    SUM(amount) AS total_sales,
    AVG(amount) AS average_sales,
    MAX(amount) AS highest_sale,
    COUNT(*) AS number_of_orders
FROM orders
GROUP BY product_id
ORDER BY total_sales DESC

Explanation:

• GROUP BY product_id creates one group for each product.
• SUM(amount) calculates total sales per product.
• AVG(amount) calculates the average order value.
• MAX(amount) identifies the highest individual order.
• COUNT(*) counts the orders in each product group.
• ORDER BY total_sales DESC lists products from highest to lowest total sales.

The query performs grouping, aggregation, and sorting in one relational operation.

SQL wildcards are special characters used with the LIKE operator to match text patterns.

• Percent wildcard (%): Matches zero or more characters.

  • name LIKE 'A%' matches names beginning with A.
  • name LIKE '%son' matches names ending with son.
  • name LIKE '%data%' matches names containing data.

• Underscore wildcard (_): Matches exactly one character.

  • code LIKE 'A_1' matches values such as AB1 or AC1.
  • name LIKE 'Jo_' matches three-character values beginning with Jo.

Wildcards are used with LIKE in Spark SQL to perform flexible text filtering. Special characters can be escaped when a literal percent or underscore is required.

The LIKE operator compares a string column with a pattern. The pattern may contain SQL wildcards.

Examples:

SELECT * FROM customers WHERE name LIKE 'S%'

Returns customers whose names begin with `S`.

sql
SELECT * FROM customers WHERE email LIKE '%@example.com'

Returns customers whose email addresses end with `@example.com`.

sql
SELECT * FROM products WHERE product_code LIKE 'AB__'

Returns product codes beginning with `AB` followed by exactly two characters.

sql
SELECT * FROM employees WHERE name NOT LIKE '%test%'

Excludes names containing the word `test`.

The percent sign matches any number of characters, while the underscore matches exactly one character. Pattern matching is useful for searching names, codes, email addresses, and other textual attributes.

RDDs, DataFrames, and Datasets are distributed data abstractions with different levels of structure and optimization.

• RDD: A distributed collection of objects. It provides low-level control and is suitable for unstructured data or custom processing.
• DataFrame: A distributed collection organized into named columns. It supports SQL operations and automatic query optimization.
• Dataset: A strongly typed distributed collection available mainly in Scala and Java. It combines type safety with Spark SQL optimization.

Comparison:

• RDDs have no schema; DataFrames and Datasets have schema information.
• RDDs provide lower-level transformations; DataFrames and Datasets provide relational operations.
• DataFrames and Datasets usually offer better performance for structured workloads.
• RDDs are useful when the operation cannot be expressed conveniently using relational APIs.
• Datasets provide compile-time type checking, while DataFrames are less strongly typed.

Spark SQL improves query performance through logical and physical optimization.

Catalyst optimizer:

• Parses SQL or DataFrame operations into a logical plan.
• Resolves tables, columns, functions, and data types.
• Applies transformations such as predicate pushdown, projection pruning, constant folding, and selection of efficient join strategies.
• Generates one or more physical execution plans and selects an appropriate plan.

Tungsten execution engine:

• Improves memory management and reduces object overhead.
• Uses compact binary data representation.
• Performs efficient CPU-level processing.
• Supports whole-stage code generation to reduce interpretation overhead.

Together, these components allow Spark SQL to execute high-level queries efficiently on distributed data.

Joins, grouping, and global sorting can require data to be redistributed between executors. This redistribution is called a shuffle.

• Joins: Rows with matching keys may need to be moved to the same partition. Large joins can consume network bandwidth and memory.
• GROUP BY: Records with the same grouping key must be brought together before aggregation. Skewed keys can cause one partition to receive much more data than others.
• ORDER BY: A global sort requires coordination across partitions and can be expensive for very large results.

Ways to improve execution:

• Filter rows early with WHERE.
• Select only required columns.
• Use broadcast joins when one table is small enough to distribute to executors.
• Repartition data using suitable keys.
• Handle data skew through salting or an appropriate partitioning strategy.
• Cache data only when it is reused and fits available memory.