Big Data File Formats

# Working with CSV in PySpark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("CSV Example").getOrCreate()

# Read CSV (slow, must infer schema)
df = spark.read.csv(
    "data/users.csv",
    header=True,
    inferSchema=True  # Scans data twice to infer types
)

# Specify schema for better performance
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("id", IntegerType(), nullable=False),
    StructField("name", StringType(), nullable=False),
    StructField("age", IntegerType(), nullable=True),
    StructField("city", StringType(), nullable=True),
    StructField("salary", IntegerType(), nullable=True)
])

df = spark.read.csv(
    "data/users.csv",
    header=True,
    schema=schema  # Much faster, no inference needed
)

# Write CSV
df.write.csv("output/users_csv", mode="overwrite", header=True)

# Performance comparison
# 1GB CSV file:
# - Read time: ~45 seconds
# - Storage: 1GB (no compression)
# - Query (SELECT name, age): Must read entire 1GB

# Working with JSON in PySpark

# Read JSON (handles nested structures automatically)
df = spark.read.json("data/users.json")

# JSON automatically infers schema from data
df.printSchema()
# root
#  |-- id: long (nullable = true)
#  |-- name: string (nullable = true)
#  |-- age: long (nullable = true)
#  |-- city: string (nullable = true)
#  |-- tags: array (nullable = true)
#  |    |-- element: string (containsNull = true)

# Access nested fields
df.select("name", df.tags[0]).show()

# Write JSON
df.write.json("output/users_json", mode="overwrite")

# Compressed JSON (much better)
df.write.json(
    "output/users_json_compressed",
    mode="overwrite",
    compression="gzip"  # or "snappy", "lz4"
)

# Performance comparison (1GB of data)
# - Read time: ~60 seconds
# - Storage: 1GB uncompressed, ~250MB with gzip
# - Very verbose: field names repeated in every record

# Working with Parquet in PySpark

# Read Parquet (very fast, schema included)
df = spark.read.parquet("data/users.parquet")

# Schema is stored in file (no inference needed)
df.printSchema()

# Write Parquet
df.write.parquet(
    "output/users_parquet",
    mode="overwrite",
    compression="snappy"  # Default, good balance
)

# Compression options
df.write.parquet("output/users_gzip", compression="gzip")      # Smaller, slower
df.write.parquet("output/users_snappy", compression="snappy")  # Balanced (default)
df.write.parquet("output/users_lz4", compression="lz4")        # Faster, larger

# Partitioning for massive datasets
df.write.parquet(
    "output/users_partitioned",
    mode="overwrite",
    partitionBy=["country", "year"]  # Physical directory structure
)
# Creates: output/users_partitioned/country=US/year=2024/*.parquet

# Column pruning (only reads needed columns)
# Query: SELECT name, age FROM users
df.select("name", "age").show()
# Parquet: Reads ONLY name and age columns
# CSV/JSON: Reads ENTIRE file

# Predicate pushdown (skips row groups)
# Query: SELECT * FROM users WHERE age > 30
df.filter("age > 30").show()
# Parquet: Checks row group metadata, skips groups where max(age) <= 30

# Performance comparison (1GB CSV → Parquet)
# - Storage: 1GB → 150MB (85% reduction)
# - Read time: 45s → 2s (22x faster)
# - Query (2 columns): 45s → 0.3s (150x faster)

# Advanced: Custom row group size
df.write.option("parquet.block.size", 256 * 1024 * 1024) \
    .parquet("output/users_large_rowgroups")  # 256MB row groups


# Reading with predicate pushdown
from pyspark.sql.functions import col

# This is extremely efficient
df = spark.read.parquet("data/users_partitioned")
filtered = df.filter(
    (col("country") == "US") &  # Partition pruning
    (col("age") > 30)           # Row group pruning
).select("name", "email")       # Column pruning

# Spark only reads relevant data:
# 1. Skips non-US partitions (partition pruning)
# 2. Skips row groups where max(age) <= 30 (predicate pushdown)
# 3. Only reads name and email columns (column pruning)
# Result: 100x+ faster than full scan