Adapt Data Ingestion

Adapt Data Ingestion

Target file: notebooks/1_data_preprocessing/data_ingestion.ipynb Target job: resources/1_data_preprocessing_job.yml

When to use

Use this skill whenever the user wants to:

• Replace the Iris dataset with their own source data.
• Add preprocessing stages (bronze → silver → gold).
• Rename the feature table or change its schema/primary key.

Step-by-step

1. Replace the data source. Remove the sklearn.datasets.load_iris() block. Replace with the customer's data source — for example:

   • Cloud storage: spark.read.format("parquet").load("s3://..."), dbutils.fs.ls + Auto Loader, etc.
   • JDBC: spark.read.format("jdbc").options(...).load().
   • REST API: pull with requests then spark.createDataFrame(...).
   • Existing Delta table: spark.read.table("...").

2. Rename the feature table. Replace iris_data with a meaningful name (e.g. customer_churn_features). Keep the three-level reference {catalog_name}.{schema_name}.<table_name> — never hardcode catalog or schema.

3. Adjust the schema. Update column casts, primary key constraints, and any column-level transformations to match the new dataset.

4. Preserve idempotency. Keep the "table exists?" guard so the notebook can safely re-run. Decide deliberately between overwrite and append for the new use case.

5. Add multi-stage preprocessing (optional). For bronze → silver → gold pipelines:

   • Add notebooks under notebooks/1_data_preprocessing/ (e.g. 1_bronze_ingestion.ipynb, 2_silver_cleaning.ipynb, 3_gold_features.ipynb).
   • Add corresponding tasks in resources/1_data_preprocessing_job.yml, wiring depends_on to enforce execution order.
   • Pass catalog_name / schema_name to each new task via base_parameters.

6. Update the job notification email. Replace your.name@address.com in resources/1_data_preprocessing_job.yml with the team's address.

Parameterization contract

Every new notebook must define widgets for catalog_name and schema_name and reference all tables as {catalog_name}.{schema_name}.<object>. See the mlops-quickstart-overview skill for the full contract.

Edge cases

• Streaming sources (Kafka, Kinesis, Auto Loader): keep ingestion in a separate notebook and configure the job task as a continuous trigger if needed.
• Large datasets: prefer Auto Loader over spark.read for incremental cloud-storage ingestion.
• Sensitive columns: apply column masks or row filters via Unity Catalog rather than dropping them in the notebook, so governance is centralized.
• Schema drift: if the source schema may evolve, enable mergeSchema (.option("mergeSchema", "true")) and document the decision in the notebook.