Learn Apache Spark - DataFrame & Spark SQL

#Introduction

DataFrame and Spark SQL are the heart of most modern Spark work. They both use the same abstraction — columnar data with a schema — so you can write logic once and execute it through the Python API, Scala API, or pure SQL.

Episode 5 covers: creating DataFrames from various file formats, running SQL queries with temporary views, understanding schema inference versus explicit schemas, and how the Catalyst optimizer produces efficient execution. This is the skill most often used day to day in the data engineering world.

#Creating DataFrames from Various Sources

#Reading CSV

CSV is the most common format for tabular data. PySpark provides the concise spark.read API:

Reading CSV with header

df = spark.read.format("csv") \
    .option("header", True) \
    .option("inferSchema", True) \
    .load("data/penjualan.csv")
df.printSchema()

df.printSchema() displays the column structure. The header and inferSchema options are a pair you'll almost always set — without inferSchema, all columns are read as strings.

#Reading JSON and Parquet

JSON is suited to semi-structured data, while Parquet is the compressed columnar format most recommended for data lakes:

Reading JSON and Parquet

json_df = spark.read.format("json").load("data/event.json")
parquet_df = spark.read.format("parquet").load("data/tabel.parquet")
parquet_df.show(5)

Because Parquet stores its schema inside the file, no additional configuration is needed — spark.read.format("parquet") understands its structure directly. This is why Parquet has become the de facto standard for data lake storage.

#SQL Queries with Spark SQL and Temporary Views

#Creating a Temporary View

To use SQL, register a DataFrame as a view:

Registering a view and running a SQL query

df.createOrReplaceTempView("penjualan")
 
hasil = spark.sql("""
    SELECT kategori, SUM(jumlah) AS total
    FROM penjualan
    GROUP BY kategori
    ORDER BY total DESC
""")
hasil.show()

df.createOrReplaceTempView("penjualan") creates a view that lives only for the duration of the session. You can write SQL queries exactly as you would against a database — you can even join views with other tables.

#Spark SQL as a Universal Language

Spark SQL isn't just an extension — the entire DataFrame API translates to the same logical plan. That means you can choose between writing df.filter(df.jumlah > 100) or spark.sql("SELECT * FROM penjualan WHERE jumlah > 100"). Both produce an identical execution plan.

df.filter(df.jumlah > 100)      ← API
spark.sql("SELECT ... WHERE ...") ← SQL

#Schema Inference and Explicit Schemas

#The Dangers of Schema Inference

inferSchema is certainly convenient, but there's a risk: Spark guesses types from the data it reads. A numeric column with one misread value can become a string, or empty values can change a column's type. For production pipelines, automatic guessing is a source of subtle bugs.

#Declaring an Explicit Schema

For full control, define a schema with explicit types:

Explicit schema with StructType

from pyspark.sql.types import StructType, StructField, StringType, IntegerType
 
schema = StructType([
    StructField("produk", StringType(), True),
    StructField("stok", IntegerType(), True),
])
 
df = spark.read.format("csv") \
    .option("header", True) \
    .schema(schema) \
    .load("data/stok.csv")

The third argument of StructField is nullable — for columns that must exist, set False so Spark validates incoming data. An explicit schema also speeds up loading because Spark doesn't need to read all the data just to guess the types.

#The Catalyst Optimizer

#How Catalyst Works

Catalyst is a rule-based optimizer that turns queries into efficient execution plans. It works through several phases: analysis, logical optimization, physical planning, and code generation.

SQL/API → logical plan → optimized plan → physical plan → RDD → result

#Optimizations You'll Feel

Some of the most impactful Catalyst optimizations:

• Predicate pushdown: filters are moved as close to the data source as possible, so fewer files are read.
• Projection pruning: columns that aren't used are never read at all.
• Constant folding: constant expressions are computed once up front.
• Join reordering: join order is optimized based on statistics.

The best way to experience Catalyst's work is to look at the execution plan:

Viewing the physical plan

df.explain("extended")

df.explain("extended") shows the logical plan and the physical plan. Making a habit of reading this output is a key skill for episode 9 on performance tuning.

#Transformations and the Expression API

#Main Transformations

The DataFrame vocabulary mirrors SQL but in method form:

A DataFrame transformation chain

hasil = df \
    .select("produk", "harga") \
    .filter(df.harga > 50000) \
    .withColumn("harga_ppn", df.harga * 1.11) \
    .groupBy("produk") \
    .agg({"harga_ppn": "avg"})

withColumn adds or replaces a column, agg takes a dictionary of aggregations. This chain is lazy — execution only happens when show(), collect(), or another action is called.

#The Expression API and Spark Functions

Spark provides built-in functions in the functions module:

Spark built-in functions

from pyspark.sql import functions as F
 
df2 = df.withColumn("bulan", F.month(df.tanggal)) \
        .withColumn("label", F.when(df.stok == 0, "habis").otherwise("tersedia"))

F.when(...).otherwise(...) is an if-else for columns. Hundreds of other functions like F.coalesce, F.window, F.row_number will keep appearing in episodes 7 and 10.

#Conclusion

Episode 5 equips you with Spark's core skill: creating DataFrames from CSV, JSON, and Parquet, querying with SQL through temporary views, controlling data types with explicit schemas, understanding the Catalyst optimizer, and chaining transformations with the expression API.

Key takeaways:

• spark.read.format(...) reads CSV, JSON, Parquet, and more.
• Temporary views bridge the DataFrame API and pure SQL.
• Explicit schemas prevent data type bugs in production.
• Catalyst performs pushdown, pruning, and reordering automatically.
• df.explain("extended") is the primary diagnostic tool.

In the next episode, episode 6, we'll discuss the Dataset API and strong typing — typed Datasets in Scala and Java, the role of encoders and compile-time type safety, the structural differences between Dataset and DataFrame, and the two-way conversion between them.