Building a sales analytics pipeline in Spark & Data Eng

Sales analytics often lives in spreadsheets and manual exports until the business scales past what humans can reasonably reconcile. Building a sales analytics pipeline in Spark & Data Eng transforms raw transactional data into reliable insights that drive revenue decisions—and the pattern holds whether you're processing millions of daily orders or thousands of monthly subscriptions.

In this post, we'll walk through a complete batch ETL pipeline that moves data from your production database through extraction, cleaning, dimensional modeling, and finally into dashboards your analytics and sales teams can trust.

Why batch ETL for sales analytics?

Before diving into the architecture, it's worth clarifying when batch makes sense. Most sales organizations don't need real-time dashboards for every transaction—daily or hourly batch runs hit the sweet spot between freshness and operational simplicity. You avoid the complexity of streaming infrastructure, gain better control over data quality gates, and make it easier to backfill and correct historical records. The pipeline is also predictable: it runs on schedule, consumes a known window of source data, and produces idempotent results.

Extract — CDC or JDBC reads from the source database

Your first challenge is reliably pulling data from the production system without impacting it. You have two main approaches: JDBC direct reads or Change Data Capture (CDC).

JDBC direct reads are simpler to start with. Spark connects to your PostgreSQL, MySQL, or Oracle instance, reads entire tables or filtered ranges, and lands them in your data lake. This works well for smaller datasets or when you're rebuilding the entire pipeline daily.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("SalesExtract") \
    .getOrCreate()

# Read sales orders with connection pooling
orders_df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://prod-db:5432/sales") \
    .option("dbtable", "public.orders") \
    .option("user", "etl_user") \
    .option("password", "***") \
    .option("numPartitions", 16) \
    .option("partitionColumn", "order_id") \
    .option("lowerBound", 1) \
    .option("upperBound", 10000000) \
    .load()

orders_df.write.mode("overwrite").parquet("s3://data-lake/raw/orders/")

For larger volumes or incremental loads, CDC approaches (Debezium, DMS, native database logs) capture only changed records. This reduces query load on production and lets you process deltas rather than full snapshots. Many teams land CDC streams into Kafka or S3 and then process them in a scheduled Spark job.

Practical recommendation: Start with JDBC if your tables fit comfortably in memory and your network is stable. Move to CDC once you're extracting multiple gigabytes daily or need sub-hour freshness.

Clean — Deduplicate and standardize currencies

Raw data is rarely perfect. Sales tables inherit duplicates from retries, late-arriving records, and system bugs. Before modeling, we standardize and deduplicate.

from pyspark.sql.window import Window
from pyspark.sql.functions import col, row_number, coalesce, upper, trim

# Load raw orders
orders_raw = spark.read.parquet("s3://data-lake/raw/orders/")

# Deduplicate by taking the latest modified record per order_id
window_spec = Window.partitionBy("order_id").orderBy(col("updated_at").desc())
orders_deduped = orders_raw \
    .withColumn("rn", row_number().over(window_spec)) \
    .filter(col("rn") == 1) \
    .drop("rn")

# Standardize currency codes and amounts
orders_clean = orders_deduped \
    .withColumn("currency_code", upper(trim(col("currency")))) \
    .filter(col("currency_code").isin(["USD", "EUR", "GBP", "JPY"])) \
    .filter(col("order_total") > 0)

# Handle nulls in critical fields
orders_clean = orders_clean.fillna({
    "customer_id": -1,
    "region": "UNKNOWN",
    "order_status": "PENDING"
})

orders_clean.write.mode("overwrite").parquet("s3://data-lake/clean/orders/")

The key patterns here are:

• Deduplication: Use window functions with row number or max timestamps to keep exactly one record per business key.
• Standardization: Uppercase, trim whitespace, and validate against a known set of values (currencies, statuses, regions).
• Null handling: Decide upfront whether nulls mean missing data, unknown category, or error—then fill them consistently.
• Data quality checks: Filter out negative amounts, impossible dates, or zero-value transactions.

Many teams add a _dq_flagged column to mark suspect rows rather than dropping them outright, so you can investigate patterns later.

Model — Fact and dimension tables

Once data is clean, build a dimensional model (star schema) that separates business events (facts) from slowly-changing attributes (dimensions). This structure makes queries fast and lets analysts reason about the data clearly.

Fact table: One row per sales transaction or order line item, with foreign keys to dimensions and numerical measures (quantity, amount, discount).

Dimension tables: Separate tables for customers, products, dates, and regions, each with a surrogate key and business attributes.

from pyspark.sql.functions import col, when, to_date, monotonically_increasing_id
from datetime import datetime

# Load cleaned data
orders_clean = spark.read.parquet("s3://data-lake/clean/orders/")
customers_raw = spark.read.parquet("s3://data-lake/clean/customers/")

# Build CUSTOMER dimension
dim_customer = customers_raw \
    .select(
        col("customer_id").alias("customer_key"),
        col("customer_id"),
        col("customer_name"),
        col("email"),
        col("country"),
        col("segment")
    ) \
    .dropDuplicates(["customer_id"])

dim_customer.write.mode("overwrite").parquet("s3://data-lake/gold/dim_customer/")

# Build DATE dimension
from pyspark.sql.functions import sequence, explode, date_format
start_date = "2020-01-01"
end_date = datetime.now().strftime("%Y-%m-%d")
dates_df = spark.select(
    explode(sequence(to_date(start_date), to_date(end_date))).alias("date_key")
).select(
    col("date_key"),
    date_format(col("date_key"), "YYYY-MM-dd").alias("date_string"),
    date_format(col("date_key"), "MMMM").alias("month_name"),
    date_format(col("date_key"), "YYYY-MM").alias("year_month"),
    date_format(col("date_key"), "YYYY-WW").alias("year_week")
)

dates_df.write.mode("overwrite").parquet("s3://data-lake/gold/dim_date/")

# Build SALES fact table
fact_sales = orders_clean \
    .join(dim_customer, orders_clean.customer_id == dim_customer.customer_id, "left") \
    .select(
        col("order_id").alias("order_key"),
        col("customer_key"),
        to_date(col("order_date")).alias("order_date_key"),
        col("product_id"),
        col("region"),
        col("order_total").alias("sales_amount"),
        col("quantity"),
        col("discount_amount"),
        col("currency_code"),
        col("order_status")
    )

fact_sales.write.mode("overwrite").parquet("s3://data-lake/gold/fact_sales/")

This dimensional structure keeps your fact table lean and your dimensions reusable across multiple pipelines. Analysts can join dimensions to facts to slice by customer segment, region, or time period without rewriting business logic.

Serve — BI dashboards on gold tables

Your gold layer (dimensional tables) is the contract with the analytics team. You expose it via a SQL layer or connect BI tools directly.

Option 1: Athena or Presto — Query Parquet files directly from S3. Simple, cost-effective, and works with most BI tools via JDBC/ODBC.

Option 2: Redshift or Snowflake — Copy gold tables into a data warehouse on a schedule. Provides better query performance and centralized governance.

Option 3: PostgreSQL or Delta Lake — Materialize gold tables in a SQL database or Delta format for real-time dashboard updates and better concurrency handling.

from pyspark.sql.functions import col, sum as spark_sum, count

# Pre-aggregate for dashboard performance
sales_by_region_month = fact_sales \
    .groupBy(col("order_date_key"), col("region")) \
    .agg(
        spark_sum(col("sales_amount")).alias("total_sales"),
        count(col("order_key")).alias("order_count"),
        (spark_sum(col("discount_amount")) / spark_sum(col("sales_amount"))).alias("avg_discount_rate")
    )

sales_by_region_month.write.mode("overwrite").parquet("s3://data-lake/gold/agg_sales_by_region_month/")

# Write to PostgreSQL for direct BI connections
sales_by_region_month.write \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://analytics-db:5432/dashboards") \
    .option("dbtable", "public.sales_by_region_month") \
    .option("user", "bi_user") \
    .option("password", "***") \
    .mode("overwrite") \
    .save()

Pre-aggregating common queries (by region, by month, by customer segment) dramatically speeds up dashboard loading. Store these aggregations in your gold layer so BI tools pull from denormalized, pre-computed tables rather than scanning billions of fact rows.

Common mistakes

• Skipping deduplication: Raw data often contains duplicates from retries or system errors. Ignoring them produces inflated revenue numbers and breaks downstream joins.
• Mixing business logic across layers: Putting currency conversion, discounts, or tax logic in the BI tool instead of the pipeline makes it hard to maintain and audit. Compute these once in the gold layer.
• No data quality gates: Launching a pipeline without checks for negative amounts, future dates, or missing customer IDs means bad data propagates to dashboards. Add assertions before writing gold tables.
• Ignoring slowly-changing dimensions: If a customer changes region, do you overwrite the dimension or track history? Decide this upfront and implement consistently (Type 1 or Type 2 slowly-changing dimension logic).
• No recovery plan for late-arriving data: Production systems often correct or re-send records hours after the initial insert. Build your pipeline to re-run the previous day or week easily, not just today.
• Over-indexing on real-time freshness: Batch ETL is simpler than streaming, but many teams over-engineer it thinking they need minute-level latency. Start with daily runs and move to hourly only if the business actually needs it.

Key takeaways

• Extract data via JDBC for small-to-medium volumes or CDC for high-volume incremental loads, keeping production database load in mind.
• Deduplicate using window functions and validate against a whitelist of valid values (currencies, statuses, regions).
• Model data as a star schema: narrow, immutable fact tables joined to dimension tables, reducing query complexity and enabling reuse.
• Pre-aggregate high-cardinality queries (by region, by month) in the gold layer so BI tools run fast without scanning raw facts.
• Expose gold tables via Athena, Snowflake, or a SQL database so analysts query without needing Spark knowledge.
• Test data quality gates (negative amounts, null keys, future dates) before writing gold tables; bad data in dashboards erodes trust.
• Make the pipeline idempotent and re-runnable; sales data often arrives late or requires correction, so you need to re-process historical windows easily.