Unit 4: Running SQL Queries Using Spark SQL

I. Orientation

Spark SQL is Apache Spark’s structured-data processing module. It combines SQL queries with Spark programs so that distributed datasets can be queried using tables, DataFrames, and Datasets while Spark’s optimizer and execution engine manage parallel computation across a cluster.

• Distributed execution: Data is divided into partitions and processed across executor nodes.
• Driver and executors: The driver builds the execution plan; executors perform tasks on partitioned data.
• Lazy evaluation: Transformations such as select() and groupBy() are planned first and executed only when an action such as show() or collect() is called.
• Schema awareness: Structured data has named columns and data types, allowing analysis and optimization before execution.
• Data abstraction: Spark SQL works with SQL tables, DataFrames, and Datasets over formats such as Parquet, JSON, CSV, and Hive tables.
• Cluster assumption: Operations should minimize unnecessary data movement, because shuffles across nodes are expensive.

II. Spark SQL

Spark SQL provides a structured interface for querying and processing large-scale data. Its central principle is to express computation through relational operations while Spark converts the request into a distributed execution plan.

A. Importance and features of Spark SQL

The importance of Spark SQL lies in its ability to make distributed data processing accessible through familiar SQL and optimized DataFrame APIs.

• SQL accessibility: Analysts can query distributed data with statements such as:

  SELECT department, AVG(salary)
  FROM employees
  GROUP BY department;

department is a grouping column, and salary is a numeric column being averaged.

• Unified processing: The same application can combine SQL, DataFrame transformations, machine-learning operations, and streaming workflows without moving data between separate systems.

• Catalyst optimization: Spark analyzes a query and applies logical optimizations such as column pruning, predicate pushdown, and constant folding before execution.

• Tungsten execution: Spark improves memory use and CPU efficiency through compact binary formats, code generation, and efficient physical execution.

• Schema enforcement: A column such as salary can be declared as DoubleType; invalid or ambiguous structure is detected earlier than with unstructured objects.

• Multiple data sources: Spark SQL can read Parquet, ORC, JSON, CSV, JDBC databases, and Hive-compatible tables. Parquet is commonly preferred because it stores schema and supports column pruning.

• Fault tolerance: If a task fails, Spark can recompute the lost partition from lineage rather than requiring the entire dataset to be reprocessed.

B. Concepts of Spark SQL

The main concepts of Spark SQL explain how data is represented, queried, optimized, and executed.

• SparkSession: SparkSession is the entry point for SQL and DataFrame operations.

  from pyspark.sql import SparkSession

  spark = SparkSession.builder.appName("SalesQuery").getOrCreate()

spark is the session object used to read data, create views, and run SQL.

• DataFrame: A DataFrame is a distributed collection of rows organized into named columns. It resembles a relational table but is partitioned across the cluster.

• Dataset: A Dataset is a typed distributed collection available mainly in Scala and Java. It combines compile-time type information with Spark SQL’s optimized execution.

• Schema: A schema describes column names and data types. For example, id: IntegerType and amount: DoubleType define the structure of a transaction record.

• Temporary view: A DataFrame can be registered as a SQL-accessible view:

  sales.createOrReplaceTempView("sales_view")
  result = spark.sql(
      "SELECT product, SUM(amount) AS total FROM sales_view GROUP BY product"
  )

The view exists within the current Spark session and does not necessarily persist data to storage.

• Logical and physical plans: Spark first creates a logical plan, optimizes it, and then selects physical operators such as scans, filters, projections, exchanges, and joins.

• Transformations and actions: filter() and select() are lazy transformations; show(), count(), and write() trigger execution.

• Schema-on-read: Data can be interpreted with a schema when it is loaded. Explicit schemas are generally more reliable than automatic inference for production pipelines.

III. Converting RDDs to DataFrames

Converting an RDD to a DataFrame adds schema information, enabling SQL expressions and query optimization. The conversion method depends on whether the RDD contains named fields or only positional values.

A. Methods to convert RDDs to DataFrames

The available methods convert ordinary Python, Scala, or Java objects into structured rows that Spark SQL can understand.

• RDD of tuples with column names: In PySpark, an RDD of tuples can be converted by supplying column names.

  rows = sc.parallelize([
      (1, "Asha", 82.5),
      (2, "Ravi", 76.0)
  ])

  students = rows.toDF(["id", "name", "marks"])

id, name, and marks become DataFrame columns, while each tuple becomes one row.

• RDD of Row objects: Row gives fields explicit names before DataFrame creation.

  from pyspark.sql import Row

  rdd = sc.parallelize([
      Row(id=1, name="Asha", marks=82.5),
      Row(id=2, name="Ravi", marks=76.0)
  ])
  students = spark.createDataFrame(rdd)

Spark can infer the schema from the values and named attributes.

• RDD with an explicit schema: An explicit StructType is appropriate when types must be controlled or the RDD contains lists and tuples.

  from pyspark.sql.types import StructType, StructField
  from pyspark.sql.types import IntegerType, StringType, DoubleType

  schema = StructType([
      StructField("id", IntegerType(), False),
      StructField("name", StringType(), True),
      StructField("marks", DoubleType(), True)
  ])

  students = spark.createDataFrame(rows, schema)

False means id cannot be null; True permits null values in the other fields.

• RDD of dictionaries: Dictionary records can be passed to createDataFrame() when field names are present, although explicit schemas remain preferable for stable pipelines.

• Reflection-based conversion: In Scala, case classes provide field names and types automatically.

  case class Student(id: Int, name: String, marks: Double)
  val students = rdd.map(x => Student(x._1, x._2, x._3)).toDF()

The case class Student defines the DataFrame schema through reflection.

• RDD conversion trade-off: Existing RDDs can be converted, but creating a DataFrame directly from the original source often gives Spark more opportunities for optimization and avoids unnecessary object conversion.

IV. Relational Query Operations

Relational operations transform rows by combining records, partitioning them into groups, sorting results, and calculating summaries. Their cost is strongly affected by partitioning and shuffle behavior.

A. Joins

Joins combine rows from two DataFrames using related column values or another join condition.

• Inner join: Returns only matching rows from both inputs.

  orders.join(customers, orders.customer_id == customers.id, "inner")

customer_id and id are join keys; unmatched customers or orders are removed.

• Left outer join: Keeps every row from the left DataFrame and adds matching right-side data. Missing right-side values become NULL.

• Right and full outer joins: A right join preserves all right-side rows; a full join preserves rows from both sides and fills unmatched columns with NULL.

• Cross join: Produces every possible pair of rows. With m rows on one side and n on the other, the result can contain m × n rows, so it should be used deliberately.

• Join strategy: Spark may use broadcast joins when one table is small enough to copy to each executor, avoiding a large shuffle.

• Duplicate columns: Joining on differently named columns can leave both key columns in the output. Selecting required columns afterward prevents ambiguity.

• SQL form:

  SELECT o.order_id, c.name
  FROM orders o
  JOIN customers c
    ON o.customer_id = c.id;

o and c are table aliases used to qualify column names.

B. GroupBy

GroupBy partitions rows according to one or more keys and prepares each group for aggregation.

• Grouping key: A key such as department identifies which rows belong together.

  employees.groupBy("department").count()

The result contains one row per department and a count column.

• Multiple keys: groupBy("department", "job_title") creates groups for each unique pair, such as (Sales, Manager).

• Shuffle cost: Rows with the same key may be transferred to the same partition. This network redistribution is a shuffle and can dominate execution time.

• Filtering groups: WHERE filters rows before grouping, whereas HAVING filters groups after aggregation.

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

Only active rows enter the groups; departments with ten or fewer workers are removed afterward.

• Null grouping values: Rows with NULL in a grouping column are generally placed in the same null group rather than discarded.

C. OrderBy

OrderBy sorts the complete query result according to one or more columns.

• Ascending and descending order:

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

salary DESC places larger salaries first; equal salaries are ordered by name ASC.

• Global ordering: orderBy() requires a global sort and commonly causes a shuffle, unlike sortWithinPartitions(), which sorts independently inside each partition.

• Null placement: Spark SQL supports explicit null ordering such as NULLS FIRST or NULLS LAST, which is important when missing values exist.

• Result limitation: Sorting before LIMIT can identify top records, but sorting a very large dataset is expensive. A query such as ORDER BY score DESC LIMIT 10 requests the ten highest scores.

• Determinism: Ordering by a non-unique column alone may leave ties in an unspecified order. Add a unique column when reproducible output is required.

D. Aggregate functions

Aggregate functions reduce multiple rows to a summary value or one value per group.

• COUNT: Counts rows or non-null values.

  SELECT COUNT(*) AS rows, COUNT(email) AS known_emails
  FROM customers;

COUNT(*) includes rows regardless of null fields; COUNT(email) excludes null email values.

• SUM: Adds numeric values, such as SUM(amount) for total sales.

• AVG: Calculates the arithmetic mean, for example AVG(marks). Null marks are normally excluded from the calculation.

• MIN and MAX: Return the smallest and largest values, respectively, such as the earliest order_date or maximum salary.

• Statistical functions: STDDEV, VARIANCE, and related functions describe dispersion rather than central tendency.

• Grouped aggregation:

  from pyspark.sql.functions import avg, sum

  employees.groupBy("department").agg(
      avg("salary").alias("average_salary"),
      sum("salary").alias("salary_total")
  )

alias() assigns readable names to computed columns.

• Null behavior: Most aggregates ignore null inputs; an aggregate over no valid numeric values may return NULL, so COALESCE() can provide a replacement value.

V. Pattern Matching in SQL

A. SQL wildcards

SQL wildcards support pattern matching in string predicates, especially with the LIKE operator.

• Percent wildcard %: Matches zero or more characters.

  SELECT *
  FROM customers
  WHERE name LIKE 'An%';

This matches names beginning with An, including An itself and longer values such as Anita.

• Underscore wildcard _: Matches exactly one character.

  SELECT *
  FROM products
  WHERE code LIKE 'A_7';

This can match AB7 or AX7, but not A007, because two characters occur between A and 7.

• Combined patterns: LIKE '%data%' finds values containing data; LIKE '___' finds values with exactly three characters.

• Negation: NOT LIKE selects values that do not satisfy a pattern.

• Escaping wildcard characters: If % or _ must be treated literally, an escape character can be specified.

  WHERE description LIKE '%10\%%' ESCAPE '\'

Here \% represents a literal percent sign rather than a wildcard.

• Case behavior: Matching behavior can depend on Spark SQL configuration, collation, and expression functions. Use functions such as lower() when a case-normalized comparison is required.

• Performance limitation: A leading wildcard such as LIKE '%error' often prevents efficient data-source filtering, while a prefix pattern such as LIKE 'error%' may allow better predicate pushdown.