NCA-ADS: Pipelines & Workflow
Automation

1. End-to-End Data Science Pipeline Design

A pipeline chains all data science steps into a reproducible, automated workflow. Each stage's output becomes the next stage's input — no manual hand-offs.

1. Data Ingestion — cudf.read_parquet() / cudf.read_csv()
2. Data Cleaning — fillna(), drop_duplicates(), type conversion
3. Feature Engineering — scaling, encoding, new feature creation
4. Train/Test Split — cuml.model_selection.train_test_split()
5. Model Training — cuml or XGBoost .fit()
6. Evaluation — metrics computation (accuracy, RMSE, AUC)
7. Model Saving — pickle / model.save_model()

• Reproducibility — same code gives same results every run
• Automation — runs end-to-end without manual steps
• Fewer manual errors — no copy-paste between steps
• Easier debugging — isolate which stage produced bad output

from cuml.pipeline import Pipeline
from cuml.preprocessing import StandardScaler
from cuml.linear_model import LogisticRegression

pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('clf', LogisticRegression())
])
pipe.fit(X_train, y_train)   # scaler fits+transforms, then clf trains
pipe.predict(X_test)         # scaler transforms only, then clf predicts

The Pipeline object chains preprocessing + model into one object. It automatically prevents data leakage by ensuring X_test is only transformed (never fit) at predict time.

2. Feature Engineering in the Pipeline Context

• Remove highly correlated features — |r| > 0.9 means two features carry redundant information
• Remove low-variance features — features with near-constant values add no signal
• Remove high-missingness features — features with >50% missing values are often unreliable
• Importance-based — use model.feature_importances_ (XGBoost) to rank and drop weak features

• Improves model generalization (fewer irrelevant signals)
• Speeds up training (fewer columns to process)
• Improves interpretability (fewer features to explain)

3. Mitigating Underfitting and Overfitting Through Pipeline Design

• L1 Regularization (Lasso) — adds sum of |weights| to loss; drives irrelevant coefficients to exactly 0 (implicit feature selection)
• L2 Regularization (Ridge) — adds sum of weights² to loss; shrinks all weights but rarely to zero
• Cross-validation within pipeline — honest performance estimate using k-fold splits
• Early stopping (XGBoost) — stop training when validation score stops improving
• Simpler model architecture — reduce tree depth, fewer estimators

• Add more features (add a feature engineering step to the pipeline)
• Use a more complex model (deeper trees, more estimators)
• Reduce regularization strength (smaller alpha/C parameter)

4. Dataset Augmentation and Integration

• Increase effective training data size without collecting new real data
• Improve model generalization to unseen examples
• Balance class distribution (address class imbalance)

# Horizontal: more rows (same columns, different time periods)
combined = cudf.concat([df_jan, df_feb, df_mar])

# Vertical: more columns (join related tables by key)
enriched = cudf.merge(customers_df, transactions_df, on='customer_id')

After integration: always check for new nulls, duplicate keys, and schema mismatches before proceeding to model training.

5. Automation and Scalability of Workflows

• Scheduling — run pipeline nightly/weekly via cron or workflow orchestrator (Apache Airflow, Prefect)
• Parameterization — pass date ranges, thresholds, and file paths as config — never hard-code
• Error handling — log failures, send alerts, retry failed steps automatically

A single GPU cannot fit all data in VRAM when the dataset exceeds GPU memory. Dask enables multi-GPU and distributed processing by partitioning data across multiple GPUs.

import dask_cudf

# Reads partitioned Parquet files lazily across multiple GPUs
ddf = dask_cudf.read_parquet('data/*.parquet')

# Operations build a task graph (nothing executes yet)
result = ddf.groupby('customer_id').agg({'amount': 'sum'})

# .compute() triggers actual computation on all GPUs
final = result.compute()

6. Building Reproducible Pipelines with RAPIDS and Dask

• Different seeds → different train/test splits → different model weights
• Different library versions → different algorithm implementations → different results
• No logging → cannot compare runs or diagnose regressions
• No data versioning → model trained on unknown data version → results unverifiable

7. Pipeline Debugging and Optimization

• Profile with nvidia-smi during run — low GPU utilization means bottleneck is elsewhere (I/O or CPU)
• Use Parquet instead of CSV for all intermediate data (10x faster reads)
• Keep data in GPU memory — avoid unnecessary .to_pandas() calls mid-pipeline
• Use cuDF categorical dtype for repeated string columns (saves memory, speeds groupby)

8. Benchmarking and Hardware Selection

import time
start = time.time()
# ... run pipeline ...
elapsed = time.time() - start
print(f"Pipeline completed in {elapsed:.1f}s")

Also use nvidia-smi during run: GPU utilization %, VRAM used/total. Low GPU% with high CPU% = I/O bottleneck. High VRAM = near memory limit.