Assemble the pieces into the pipeline that actually feeds a model: ingest to the lake, transform with Spark, build a feature table, and schedule it — the MLOps data backbone.
Why: everything in this course composes into one flow that turns raw sources into model-ready features — ingest to the lake, transform with Spark, write a curated feature table — and it is the data half of every ML system. When: this pipeline runs on a schedule (or on new data) to keep features fresh for training and serving. Where: it hands off directly to the training pipeline and feature store from the MLOps roadmap.
sources ─► INGEST ─► LAKE (raw Parquet) ─► SPARK transform ─► FEATURE TABLE
(DB/API/ (batch or (partitioned by (clean, join, (curated,
Kafka) stream) date) aggregate) ML-ready)
│
training + servingWhy: the heart of the pipeline is a single, version-controlled transform that reads raw lake data and produces the feature table — keeping it as one function ensures training and serving compute features identically (no skew). When: run it in batch for training data; reuse the same code path online where possible. Where: this is the "same transformation code in both paths" rule from the Feature Pipeline checklist.
from pyspark.sql import functions as F
def build_features(spark, run_date: str):
events = spark.read.parquet(f"s3://lake/events/dt={run_date}/")
features = (events
.filter(F.col("event") == "purchase")
.groupBy("user_id")
.agg(
F.count("*").alias("purchase_count"),
F.sum("amount").alias("total_spent"),
F.avg("amount").alias("avg_order_value"),
))
features.write.mode("overwrite").parquet(
f"s3://warehouse/features/user_purchases/dt={run_date}/")
return featuresWhy: a data pipeline is only useful when it runs reliably and on time, so it is orchestrated — scheduled, retried, and monitored — and wired to the rest of the ML stack. When: run it as an Airflow DAG that lands data, builds features, and triggers training. Where: this closes the loop with the Orchestration and MLflow courses — the orchestrator runs this pipeline, then kicks off tracked training on its output.
# An Airflow DAG that runs this data pipeline, then triggers training.
from airflow.decorators import dag, task
import pendulum
@dag(start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
schedule="@daily", catchup=False)
def feature_pipeline():
@task
def ingest(ds=None): ... # pull new data to the lake for the run date
@task
def transform(ds=None): ... # run build_features(spark, ds)
@task
def trigger_training(ds=None): ... # kick off the MLflow-tracked training
trigger_training(transform(ingest()))
feature_pipeline()