Reading and writing data for Spark & Data Eng developers

Reading and writing data for Spark & data eng developers

Getting data in and out of your Spark pipelines efficiently is often where the real performance battles are won or lost. The format you choose, how you define your schema, and which connector you use can mean the difference between a job that completes in minutes and one that grinds through hours of unnecessary I/O. We'll walk through the core patterns and formats that teams use across production systems—from analytics warehouses to real-time ingestion pipelines.

Parquet is the default for analytics

If you're writing analytical data in Spark, Parquet should be your starting point. It's columnar, compressed, and Spark understands its schema natively. When you write a DataFrame to Parquet, Spark includes schema information in the file metadata, so downstream readers don't have to guess or specify types manually.

The real wins come from Parquet's compression and column pruning. When you query only a subset of columns, Spark reads just those columns from disk—not the entire row. On a 100-column table where you're selecting five columns, this is transformative for both speed and cost.

from pyspark.sql import SparkSession

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

# Write DataFrame to Parquet
df.write.mode("overwrite").parquet("s3://my-bucket/analytics/events/")

# Read it back—schema is preserved
df_read = spark.read.parquet("s3://my-bucket/analytics/events/")
print(df_read.schema)

In production, we partition Parquet tables by date or region to enable efficient filtering. Spark's partitioning support means when you filter by date = '2024-01-15', only that partition's files are read.

# Write with partitioning
df.write \
  .mode("overwrite") \
  .partitionBy("date", "region") \
  .parquet("s3://my-bucket/analytics/events/")

# Reading specific partition is automatic
df_jan = spark.read.parquet("s3://my-bucket/analytics/events/") \
  .filter("date = '2024-01-15' AND region = 'us-west'")

One practical note: be cautious with small partition sizes. If you partition by hour and generate 24 files per day, that's manageable. If you partition by user ID with millions of values, you'll create millions of small files, and Spark's overhead in managing them will crush your performance. Aim for partition sizes in the 128MB–1GB range.

CSV and JSON for ingestion

CSV and JSON are the lingua franca of data ingestion. Rarely does raw data arrive as Parquet; it comes from APIs, logs, database exports, and legacy systems in JSON or CSV form. Spark has solid readers for both, but they require care.

CSV is lightweight and familiar but ambiguous without a schema. A field that looks like 00123 could be a string or number. Spark has to infer, and inference can guess wrong, especially with sparse data where a column is null for the first thousand rows then suddenly contains numbers.

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

# Always define schema explicitly for raw files
csv_schema = StructType([
    StructField("user_id", StringType(), False),
    StructField("event_type", StringType(), False),
    StructField("amount", IntegerType(), True),
    StructField("timestamp", TimestampType(), False)
])

df_csv = spark.read \
  .schema(csv_schema) \
  .option("header", "true") \
  .option("mode", "FAILFAST") \
  .csv("s3://my-bucket/raw/events.csv")

print(f"Read {df_csv.count()} rows")

The FAILFAST mode is crucial in production: if a row doesn't match the schema, the job fails immediately rather than silently dropping or corrupting data. Swap to PERMISSIVE only during exploration, and even then, capture malformed records in a separate column.

JSON requires similar care. A single JSON file may contain objects with different field sets (one record has email, another doesn't). Spark's JSON reader is forgiving but will create nulls for missing fields, which is usually what you want.

json_schema = StructType([
    StructField("user_id", StringType(), False),
    StructField("properties", StringType(), True),
    StructField("timestamp", TimestampType(), False)
])

df_json = spark.read \
  .schema(json_schema) \
  .option("mode", "FAILFAST") \
  .json("s3://my-bucket/raw/api-logs/")

# JSON paths let you extract nested fields
df_nested = df_json.selectExpr(
    "user_id",
    "get_json_object(properties, '$.event_type') as event_type",
    "timestamp"
)

A practical pattern in teams we work with: write CSV and JSON to a raw zone, validate and transform to Parquet in a staging zone, then move validated Parquet to the analytics zone. This separation means bad data doesn't corrupt your analytics warehouse.

JDBC for databases

When you need to read or write relational databases—PostgreSQL, MySQL, SQL Server—JDBC is the standard connector. Spark ships with JDBC support, and it handles partitioned reads across a cluster efficiently.

For reads, JDBC can parallelize by splitting the query range. If you're reading a users table with IDs 1–10,000,000, Spark can spawn multiple JDBC connections, each fetching a partition of that ID range.

jdbc_url = "jdbc:postgresql://prod-db.internal:5432/analytics"
jdbc_user = "spark_reader"
# Password from secrets manager, not hardcoded
jdbc_password = get_secret("spark_db_password")

df_pg = spark.read \
  .format("jdbc") \
  .option("url", jdbc_url) \
  .option("dbtable", "public.events") \
  .option("user", jdbc_user) \
  .option("password", jdbc_password) \
  .option("numPartitions", 8) \
  .option("partitionColumn", "event_id") \
  .option("lowerBound", 1) \
  .option("upperBound", 10000000) \
  .load()

print(f"Read {df_pg.count()} rows across 8 partitions")

The numPartitions, partitionColumn, lowerBound, and upperBound options are crucial for performance. Without them, Spark reads everything serially on one connection. With them, you distribute the load across your cluster.

For writes back to a database, be aware that Spark will insert data in parallel, which can overwhelm a small database or violate unique constraints if you're not careful. Test with a small dataset first, and ensure your database has adequate connection pool size.

# Write aggregated results back to database
df_aggregated.write \
  .format("jdbc") \
  .option("url", jdbc_url) \
  .option("dbtable", "public.daily_summary") \
  .option("user", jdbc_user) \
  .option("password", jdbc_password) \
  .mode("append") \
  .save()

In high-frequency pipelines, avoid writing to the database on every micro-batch. Instead, buffer results to Parquet, then batch-load into the database on a schedule (hourly, daily). This protects your database from connection exhaustion and keeps insert overhead minimal.

Always define schemas for raw files

This bears repeating because we see it cause subtle, expensive bugs. When you read a CSV or JSON file without an explicit schema, Spark samples the file to infer types. This inference is lossy and fragile.

Consider a CSV where the first 10,000 rows have null values in a revenue column, and row 10,001 has 123.45. If Spark samples only the first 1,000 rows, it infers revenue as null or string. Later, when the real data arrives, type mismatches or silent truncation may occur.

A schema is a contract: it documents what the data should look like, and it enforces that contract at read time. If data doesn't conform, you catch it immediately rather than discovering subtle data quality issues downstream.

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

# Schema as a contract for raw data
raw_events_schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("user_id", StringType(), False),
    StructField("event_type", StringType(), False),
    StructField("revenue", DoubleType(), True),
    StructField("currency", StringType(), True),
    StructField("event_date", DateType(), False),
    StructField("created_at", TimestampType(), False)
])

df = spark.read \
  .schema(raw_events_schema) \
  .option("mode", "FAILFAST") \
  .csv("s3://raw/events/")

# Now you can rely on types
df = df.filter(df.revenue > 0)  # Works safely
df = df.withColumn("revenue_eur", df.revenue * 0.92)  # No type uncertainty

Store schemas in a version-controlled location—a Python module, a JSON file, or a schema registry. Teams we've worked with often share schemas between producers (the system that generates CSV) and consumers (Spark). This alignment prevents weeks of debugging.

Common mistakes

• Inferring schemas in production: Tempting for quick prototyping, dangerous in production. Schema inference is non-deterministic and can change as data patterns shift.
• Not partitioning Parquet writes: Writing millions of rows to a single Parquet file creates a bottleneck. Partition by date, region, or another natural key to keep file sizes reasonable.
• Reading JDBC without partitioning: If you omit partitioning options, Spark fetches all data on a single connection. For multi-million-row tables, this is slow and risks overwhelming your database.
• Trusting CSV inference for numeric IDs: User IDs or product IDs that start with zeros are read as strings via inference but may be read as integers by schema, losing leading zeros. Always specify StringType for IDs.
• Writing CSV with no schema validation: A DataFrame read from an unvalidated CSV can have nulls or type mismatches. Validate against a schema before writing downstream.
• Using PERMISSIVE mode for production ingestion: This silently drops malformed rows. You lose data without knowing. Use FAILFAST to catch errors early.

Key takeaways

• Parquet is the standard for analytics storage: columnar, compressed, schema-aware, and efficient for column pruning and partition filtering.
• CSV and JSON are ingestion formats; always define explicit schemas with FAILFAST mode to catch data quality issues immediately.
• JDBC reads benefit dramatically from partitioning options (partitionColumn, lowerBound, upperBound) to distribute load across the cluster.
• Schema definition is a contract between producer and consumer; store schemas in version control and enforce them at read time.
• Separate raw data (CSV/JSON), staged data (validated Parquet), and analytics data (final Parquet) to isolate bad data from your warehouse.
• For high-frequency writes to databases, buffer to Parquet and batch-load rather than writing on every micro-batch to avoid connection exhaustion.