🤖 MLLIB
HOME

🧠 Fundamentals

🎯 What is Distributed ML?

Distributed machine learning trains models across multiple machines to handle datasets and models too large for a single machine. MLlib is Apache Spark's distributed ML library for classical algorithms, while frameworks like Horovod, PyTorch DDP, and TensorFlow Distributed handle distributed deep learning.

Why Distributed ML Matters

Scale: Billions of records. Speed: Parallel training. Memory: Beyond single machine. Models: Large neural networks. Production: End-to-end pipelines. Cost: Commodity hardware.

📌 Scale📌 Parallelism📌 MLlib📌 Distributed Training
# MLlib & Distributed ML — the big picture # Scaling machine learning to big data. # CONCEPTUAL OVERVIEW print(""" === Distributed ML === What it is: - Train models across machines - Handle big data - Handle large models - Parallel computation Two main types: 1. Distributed classical ML (MLlib) 2. Distributed deep learning (Horovod, DDP) Approaches: - Data parallelism - Model parallelism - Hybrid Frameworks: - MLlib (Spark) - Horovod (Uber) - PyTorch DDP - TensorFlow Distributed - Ray - Dask-ML """) # WHY IT MATTERS print("=== Why It Matters ===") print(""" Challenges: - Data too large - Model too large - Training too slow - Memory limits Distributed ML solves: - Scale to billions of records - Train large models - Faster training - Use commodity hardware - Production-ready """) # REAL-WORLD print("=== Real-World Scale ===") print(""" Companies using distributed ML: - Google (TPU pods) - Meta (PyTorch DDP) - Uber (Horovod) - OpenAI (large models) - Netflix (recommendations) - LinkedIn (ranking) Scales: - Billions of records - Trillions of parameters - Thousands of GPUs - Petabytes of data """)

📊 Distributed ML

🎯 Scaling Machine Learning

Distributed ML splits work across machines. The main strategies are data parallelism, model parallelism, and hybrid approaches.

Distributed ML

Data: Split data. Model: Split model. Hybrid: Both. Sync: Gradients. Async: Updates. Framework: MLlib, Horovod.

📊 Data📊 Model📊 Hybrid📊 Sync/Async
# Distributed ML print("=== Main Approaches ===") print(""" 1. DATA PARALLELISM - Split data across workers - Each trains a model copy - Aggregate gradients - Most common 2. MODEL PARALLELISM - Split model across workers - Each computes a portion - Aggregate activations - For large models 3. HYBRID - Both data and model parallelism - Used for largest models - Complex 4. PIPELINE PARALLELISM - Split model into stages - Different workers per stage - Micro-batching """) print("\n=== Synchronization ===") print(""" 1. SYNCHRONOUS - All workers update together - Wait for slowest - Deterministic - Simpler 2. ASYNCHRONOUS - Workers update independently - No waiting - Faster but less stable - Stale gradients 3. STALE SYNCHRONOUS - Hybrid - Bounded staleness - Balance """) print("\n=== Communication ===") print(""" Strategies: - AllReduce (ring) - Parameter server - Tree reduction - Hierarchical Costs: - Network bandwidth - Latency - Synchronization overhead Optimization: - Gradient compression - Quantization - Overlapping computation - Hierarchical aggregation """) print("\n=== Architecture ===") print(""" Data Parallelism: ┌─────────┐ ┌─────────┐ ┌─────────┐ │Worker 1 │ │Worker 2 │ │Worker 3 │ │(GPU) │ │(GPU) │ │(GPU) │ │ │ │ │ │ │ │ Data 1 │ │ Data 2 │ │ Data 3 │ │ Model │ │ Model │ │ Model │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ └────────────┼────────────┘ │ ▼ ┌───────────┐ │ AllReduce │ │ (Sync) │ └───────────┘ """) print("\n=== When to Use Which ===") print(""" DATA PARALLELISM: ✓ Large datasets ✓ Model fits on 1 GPU ✓ Most common MODEL PARALLELISM: ✓ Large models (>1 GPU) ✓ Transformers, LLMs ✓ Complex HYBRID: ✓ Massive scale ✓ Largest models ✓ Frontier research """)

🔀 Data Parallelism

🎯 Split the Data

Data parallelism is the most common distributed ML strategy. Each worker gets a copy of the model and a shard of the data. Gradients are aggregated after each step.

Data Parallelism

Split: Data. Replicate: Model. Train: Locally. Aggregate: Gradients. Sync: AllReduce. Update: All workers.

🔀 Split Data🔀 Replicate Model🔀 Aggregate🔀 Update
# Data Parallelism print("=== How It Works ===") print(""" 1. Split data: - Worker 1: batch 1 - Worker 2: batch 2 - Worker 3: batch 3 2. Forward pass: - Each worker computes loss 3. Backward pass: - Each worker computes gradients 4. AllReduce: - Average gradients - All workers get same 5. Update: - Each worker updates model - Models stay in sync """) print("\n=== PyTorch DDP ===") print(""" import torch import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDP # Init dist.init_process_group("nccl") rank = dist.get_rank() device = rank % torch.cuda.device_count() torch.cuda.set_device(device) # Model model = MyModel().to(device) model = DDP(model, device_ids=[device]) # Data dataset = MyDataset() sampler = DistributedSampler(dataset) loader = DataLoader(dataset, sampler=sampler, batch_size=32) # Train optimizer = torch.optim.Adam(model.parameters()) for epoch in range(epochs): sampler.set_epoch(epoch) for batch in loader: optimizer.zero_grad() loss = model(batch) loss.backward() # AllReduce happens here optimizer.step() """) print("\n=== Horovod ===") print(""" import horovod.torch as hvd # Init hvd.init() torch.cuda.set_device(hvd.local_rank()) # Model model = MyModel().cuda() optimizer = torch.optim.Adam(model.parameters()) optimizer = hvd.DistributedOptimizer(optimizer) # Broadcast initial state hvd.broadcast_parameters(model.state_dict(), root_rank=0) # Data train_sampler = torch.utils.data.distributed.DistributedSampler( dataset, num_replicas=hvd.size(), rank=hvd.rank() ) # Train for epoch in range(epochs): train_sampler.set_epoch(epoch) for batch in loader: optimizer.zero_grad() loss = model(batch) loss.backward() # AllReduce optimizer.step() """) print("\n=== TensorFlow Distributed ===") print(""" import tensorflow as tf # Strategy strategy = tf.distribute.MirroredStrategy() with strategy.scope(): model = create_model() model.compile(optimizer='adam', loss='sparse_categorical_crossentropy') # Data train_dist = strategy.experimental_distribute_dataset(train_dataset) # Train model.fit(train_dist, epochs=10) """) print("\n=== Key Concepts ===") print(""" 1. EFFECTIVE BATCH SIZE - Global batch = local × workers - Larger = more stable - Adjust learning rate 2. GRADIENT AGGREGATION - AllReduce - Average - Sum 3. SYNCHRONIZATION - Every step - Bucket gradients - Overlap with compute 4. SCALING - Linear speedup ideal - Communication overhead - Diminishing returns """) print("\n=== Challenges ===") print(""" 1. COMMUNICATION - Network bandwidth - Latency - AllReduce cost 2. SYNCHRONIZATION - Wait for slowest - Stragglers 3. EFFECTIVE BATCH SIZE - Larger batch = less updates - Learning rate scaling - Generalization gap 4. FAULT TOLERANCE - Worker failure - Checkpointing - Restart """)

🧩 Model Parallelism

🎯 Split the Model

Model parallelism splits the model itself across workers. Each worker holds a portion of the model and computes its part. Used when models are too large for a single GPU.

Model Parallelism

Split: Model. Each: Part of model. Pipeline: Data flows. Communication: Activations. Complex: Harder. LLMs: Common.

🧩 Split Model🧩 Pipeline🧩 Activations🧩 LLMs
# Model Parallelism print("=== How It Works ===") print(""" 1. Split model: - Worker 1: layers 1-10 - Worker 2: layers 11-20 - Worker 3: layers 21-30 2. Forward pass: - Worker 1 → Worker 2 → Worker 3 - Activations passed 3. Backward pass: - Worker 3 → Worker 2 → Worker 1 - Gradients passed 4. Update: - Each worker updates its layers """) print("\n=== Pipeline Parallelism ===") print(""" Pipeline: - Split model into stages - Micro-batches flow through - Overlap compute Example: Worker 1: Stage 1 Worker 2: Stage 2 Worker 3: Stage 3 Micro-batch 1: W1 → W2 → W3 Micro-batch 2: W1 → W2 → W3 Micro-batch 3: W1 → W2 → W3 All workers busy """) print("\n=== PyTorch Pipeline ===") print(""" from torch.distributed.pipeline.sync import Pipe # Split model class MyModel(nn.Module): def __init__(self): super().__init__() self.layers = nn.Sequential( nn.Linear(100, 200), nn.ReLU(), nn.Linear(200, 100), nn.ReLU(), nn.Linear(100, 10), ) # Split across devices model = MyModel() model = Pipe(model, balance=[2, 2, 1], chunks=8) # balance: layers per device # chunks: micro-batches # Train output = model(input) loss = criterion(output, target) loss.backward() """) print("\n=== Megatron-LM ===") print(""" Megatron-LM (NVIDIA): - Model parallelism for transformers - Tensor parallelism - Pipeline parallelism - Data parallelism - Combined Used for: - GPT-3 scale - LLMs - Trillion-parameter models Key techniques: - Tensor parallelism (within layer) - Pipeline parallelism (across layers) - Data parallelism (across batches) """) print("\n=== Tensor Parallelism ===") print(""" Tensor parallelism: - Split individual tensors - Matrix multiplication split - Column parallel - Row parallel Example: Weight matrix: 1000x1000 Split into: 2x (500x1000) Each worker: 500x1000 Result: concatenated """) print("\n=== Comparison ===") print(f"{'Aspect':<25} {'Data':<25} {'Model':<25}") print("-" * 75) print(f"{'Split':<25} {'Data':<25} {'Model':<25}") print(f"{'Replicate':<25} {'Model':<25} {'Data':<25}") print(f"{'Communication':<25} {'Gradients':<25} {'Activations':<25}") print(f"{'Complexity':<25} {'Simple':<25} {'Complex':<25}") print(f"{'Use case':<25} {'Large data':<25} {'Large model':<25}") print(f"{'Common':<25} {'Yes':<25} {'For LLMs':<25}")

🌐 Parameter Server

🎯 Centralized Gradient Aggregation

The parameter server architecture uses a central server (or group of servers) to store model parameters. Workers compute gradients and push them to the server, which aggregates and updates.

Parameter Server

Server: Parameters. Workers: Gradients. Push: Gradients. Pull: Parameters. Async: Possible. Scale: Large.

🌐 Server🌐 Workers🌐 Push/Pull🌐 Async
# Parameter Server print("=== Architecture ===") print(""" ┌─────────────┐ │ Parameter │ │ Server │ │ (Parameters)│ └──────┬──────┘ │ ┌──────┼──────┐ │ │ │ ▼ ▼ ▼ ┌────┐ ┌────┐ ┌────┐ │W 1 │ │W 2 │ │W 3 │ │ │ │ │ │ │ │Grad│ │Grad│ │Grad│ └────┘ └────┘ └────┘ Flow: 1. Workers pull parameters 2. Compute gradients 3. Push gradients to server 4. Server aggregates 5. Updates parameters 6. Repeat """) print("\n=== Synchronous vs Asynchronous ===") print(""" Synchronous: - All workers push - Server aggregates - Broadcasts new params - Wait for all - Deterministic Asynchronous: - Workers push anytime - Server updates immediately - No waiting - Faster - Stale gradients Stale synchronous: - Bound staleness - Balance """) print("\n=== Implementations ===") print(""" 1. DISTBELIEF (Google) - Original - Async SGD - Large scale 2. PETUUM - Open source - Parameter server - Async 3. MXNET - Parameter server - KVStore - Distributed 4. TENSORFLOW - ParameterServerStrategy - Sync/async - Built-in 5. RAY - Actor-based - Parameter server - Flexible """) print("\n=== TensorFlow Parameter Server ===") print(""" import tensorflow as tf # Cluster cluster_resolver = tf.distribute.cluster_resolver.TFConfigClusterResolver() # Strategy strategy = tf.distribute.experimental.ParameterServerStrategy( cluster_resolver ) with strategy.scope(): model = create_model() model.compile(optimizer='adam', loss='mse') # Data dataset = strategy.experimental_distribute_dataset(train_ds) # Train model.fit(dataset, epochs=10) """) print("\n=== Pros & Cons ===") print(""" PROS: ✓ Scalable ✓ Async possible ✓ Flexible ✓ Fault tolerant CONS: ✗ Server bottleneck ✗ Network overhead ✗ Stale gradients (async) ✗ Complex """) print("\n=== When to Use ===") print(""" PARAMETER SERVER: ✓ Very large models ✓ Async training OK ✓ Heterogeneous workers ✓ Fault tolerance needed ALLREDUCE: ✓ Synchronous ✓ Small-medium models ✓ Homogeneous workers ✓ Simpler Modern trend: - AllReduce preferred (Horovod, NCCL) - Parameter server for specific cases """)

🔄 AllReduce

🎯 Efficient Gradient Aggregation

AllReduce is a collective communication operation that aggregates values across all workers and distributes the result to all. It's the foundation of modern distributed training (Horovod, NCCL).

AllReduce

Collective: All workers. Aggregate: Sum/avg. Distribute: Result. Ring: Efficient. NCCL: GPU. Gloo: CPU.

🔄 Collective🔄 Ring🔄 NCCL🔄 Efficient
# AllReduce print("=== What is AllReduce? ===") print(""" AllReduce: - Collective operation - All workers participate - Aggregate values - Distribute result - All get same Example: Worker 1: [1, 2, 3] Worker 2: [4, 5, 6] Worker 3: [7, 8, 9] AllReduce (sum): All workers: [12, 15, 18] AllReduce (avg): All workers: [4, 5, 6] """) print("\n=== Ring AllReduce ===") print(""" Ring topology: - Workers in a ring - Each sends to next - Each receives from prev Phases: 1. Scatter-reduce: - Reduce chunks - N-1 steps 2. All-gather: - Distribute results - N-1 steps Total: 2(N-1) steps Optimal: bandwidth-efficient """) print("\n=== Implementations ===") print(""" 1. NCCL (NVIDIA) - GPU-optimized - Ring + tree - Fast - Default for PyTorch DDP 2. GLOO (Facebook) - CPU - Also GPU - Flexible - Good for CPU 3. MPI - Classic - CPU - Many implementations 4. HOROVOD - Wraps NCCL/Gloo/MPI - Simple API - Multi-framework """) print("\n=== Horovod Example ===") print(""" import horovod.torch as hvd # Init hvd.init() # Model model = MyModel().cuda() optimizer = torch.optim.Adam(model.parameters()) # Wrap with Horovod optimizer = hvd.DistributedOptimizer( optimizer, named_parameters=model.named_parameters() ) # Broadcast initial hvd.broadcast_parameters(model.state_dict(), root_rank=0) # Train (AllReduce happens automatically) for batch in loader: optimizer.zero_grad() loss = model(batch) loss.backward() optimizer.step() """) print("\n=== NCCL Example ===") print(""" # PyTorch uses NCCL by default for GPU import torch.distributed as dist dist.init_process_group("nccl") # AllReduce directly tensor = torch.tensor([1.0, 2.0, 3.0]).cuda() dist.all_reduce(tensor, op=dist.ReduceOp.SUM) # All workers now have sum # AllGather output = [torch.zeros(3).cuda() for _ in range(world_size)] dist.all_gather(output, tensor) """) print("\n=== Comparison ===") print(f"{'Aspect':<25} {'Ring':<25} {'Tree':<25}") print("-" * 75) print(f"{'Bandwidth':<25} {'Optimal':<25} {'Good':<25}") print(f"{'Latency':<25} {'Higher':<25} {'Lower':<25}") print(f"{'Best for':<25} {'Large data':<25} {'Small data':<25}") print(f"{'Fault tolerance':<25} {'Lower':<25} {'Higher':<25}") print("\n=== When to Use ===") print(""" ALLREDUCE: ✓ Synchronous training ✓ Data parallelism ✓ Small-medium models ✓ Homogeneous hardware ✓ Simple HOROVOD: ✓ Multi-framework ✓ Easy integration ✓ Production NCCL: ✓ GPU training ✓ Fastest ✓ Default in PyTorch """)

⚡ Spark MLlib

🎯 Distributed Classical ML

MLlib is Apache Spark's distributed machine learning library. It provides scalable implementations of classical ML algorithms that run on Spark clusters.

MLlib

Distributed: Spark. Algorithms: Classical. Pipelines: ML. DataFrames: Based. Scale: Billions. Integration: Spark.

⚡ Spark⚡ Classical ML⚡ Pipelines⚡ Scale
# Spark MLlib print("=== What is MLlib? ===") print(""" MLlib: - Spark's ML library - Distributed - Classical algorithms - DataFrame-based (ml) - RDD-based (mllib, legacy) - Scale to billions Two APIs: - spark.ml: DataFrame-based (recommended) - spark.mllib: RDD-based (legacy) Algorithms: - Classification - Regression - Clustering - Collaborative filtering - Dimensionality reduction - Feature extraction - Pipelines """) print("\n=== Why MLlib? ===") print(""" 1. SCALE - Billions of records - Distributed - Spark clusters 2. INTEGRATION - Spark SQL - DataFrames - Streaming - GraphX 3. PIPELINES - End-to-end - Feature + model - Cross-validation 4. PRODUCTION - Save/load - Deploy - Streaming 5. OPEN SOURCE - Apache 2.0 - Community - Mature """) print("\n=== Example: Classification ===") print(""" from pyspark.ml import Pipeline from pyspark.ml.classification import LogisticRegression from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.sql import SparkSession # Spark spark = SparkSession.builder.appName("MLlib").getOrCreate() # Load data df = spark.read.csv("data.csv", header=True, inferSchema=True) # Feature engineering assembler = VectorAssembler( inputCols=["age", "salary", "tenure"], outputCol="features" ) scaler = StandardScaler( inputCol="features", outputCol="scaled_features" ) # Model lr = LogisticRegression( featuresCol="scaled_features", labelCol="label", maxIter=100, regParam=0.01 ) # Pipeline pipeline = Pipeline(stages=[assembler, scaler, lr]) # Train/test split train, test = df.randomSplit([0.8, 0.2], seed=42) # Train model = pipeline.fit(train) # Predict predictions = model.transform(test) predictions.select("label", "prediction", "probability").show(5) # Evaluate evaluator = BinaryClassificationEvaluator(labelCol="label") auc = evaluator.evaluate(predictions) print(f"AUC: {auc}") # Save model.save("models/lr_model") """) print("\n=== Example: Clustering ===") print(""" from pyspark.ml.clustering import KMeans from pyspark.ml.feature import VectorAssembler # Data df = spark.read.parquet("features.parquet") # Features assembler = VectorAssembler( inputCols=["feature1", "feature2", "feature3"], outputCol="features" ) df = assembler.transform(df) # KMeans kmeans = KMeans(k=5, seed=42, featuresCol="features") model = kmeans.fit(df) # Predict predictions = model.transform(df) predictions.select("features", "prediction").show(5) # Centers print("Centers:") for center in model.clusterCenters(): print(center) """) print("\n=== Example: Recommendation (ALS) ===") print(""" from pyspark.ml.recommendation import ALS from pyspark.ml.evaluation import RegressionEvaluator # Data ratings = spark.read.csv("ratings.csv", header=True, inferSchema=True) # ALS als = ALS( userCol="user_id", itemCol="item_id", ratingCol="rating", rank=10, maxIter=10, regParam=0.1, coldStartStrategy="drop" ) # Split train, test = ratings.randomSplit([0.8, 0.2], seed=42) # Train model = als.fit(train) # Predict predictions = model.transform(test) # Evaluate evaluator = RegressionEvaluator( metricName="rmse", labelCol="rating", predictionCol="prediction" ) rmse = evaluator.evaluate(predictions) print(f"RMSE: {rmse}") # Recommend for users user_recs = model.recommendForAllUsers(10) user_recs.show(5, truncate=False) """) print("\n=== Pipelines ===") print(""" Pipeline: - Sequence of stages - Transformers + Estimators - Fit + transform - Cross-validation - Save/load Example: pipeline = Pipeline(stages=[ assembler, scaler, lr ]) # Cross-validation param_grid = ParamGridBuilder() \\ .addGrid(lr.regParam, [0.01, 0.1, 1.0]) \\ .addGrid(lr.maxIter, [50, 100]) \\ .build() cv = CrossValidator( estimator=pipeline, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5 ) cv_model = cv.fit(train) best_model = cv_model.bestModel """)

📚 MLlib Algorithms

🎯 What MLlib Offers

MLlib provides a comprehensive set of distributed algorithms for classification, regression, clustering, recommendation, and more.

Algorithms

Classification: LR, DT, RF, GBT, SVM, NB. Regression: Linear, GLM, DT. Clustering: K-means, LDA. Recommendation: ALS. Dim reduction: PCA, SVD. Feature: VectorAssembler.

📚 Classification📚 Regression📚 Clustering📚 Recommendation
# MLlib Algorithms print("=== Classification ===") print(""" - LogisticRegression - DecisionTreeClassifier - RandomForestClassifier - GBTClassifier - LinearSVC - NaiveBayes - MultilayerPerceptronClassifier - OneVsRest Example: from pyspark.ml.classification import RandomForestClassifier rf = RandomForestClassifier( numTrees=100, maxDepth=10, featuresCol="features", labelCol="label" ) """) print("\n=== Regression ===") print(""" - LinearRegression - GeneralizedLinearRegression - DecisionTreeRegressor - RandomForestRegressor - GBTRegressor - IsotonicRegression - AFTSurvivalRegression Example: from pyspark.ml.regression import GBTRegressor gbt = GBTRegressor( maxIter=100, maxDepth=5, featuresCol="features", labelCol="label" ) """) print("\n=== Clustering ===") print(""" - KMeans - BisectingKMeans - GaussianMixture - LDA (topic modeling) - PowerIterationClustering Example: from pyspark.ml.clustering import BisectingKMeans bkm = BisectingKMeans( k=5, featuresCol="features" ) """) print("\n=== Collaborative Filtering ===") print(""" - ALS (Alternating Least Squares) Example: from pyspark.ml.recommendation import ALS als = ALS( userCol="user", itemCol="item", ratingCol="rating", rank=10 ) """) print("\n=== Dimensionality Reduction ===") print(""" - PCA - SVD - FeatureHasher Example: from pyspark.ml.feature import PCA pca = PCA( k=10, inputCol="features", outputCol="pca_features" ) """) print("\n=== Feature Engineering ===") print(""" - VectorAssembler - VectorIndexer - StandardScaler - MinMaxScaler - MaxAbsScaler - Normalizer - OneHotEncoder - StringIndexer - IndexToString - Tokenizer - StopWordsRemover - CountVectorizer - TF-IDF - Word2Vec - Bucketizer - QuantileDiscretizer - Imputer - Binarizer - PCA - ChiSqSelector """) print("\n=== Statistics ===") print(""" - Correlation - ChiSquareTest - Summarizer - KolmogorovSmirnovTest """) print("\n=== Tuning ===") print(""" - CrossValidator - TrainValidationSplit - ParamGridBuilder - Evaluator (binary, multiclass, regression, clustering) """) print("\n=== Algorithm Comparison ===") print(f"{'Category':<20} {'Algorithms':<50}") print("-" * 75) print(f"{'Classification':<20} {'LR, DT, RF, GBT, SVM, NB, MLP':<50}") print(f"{'Regression':<20} {'Linear, GLM, DT, RF, GBT, AFT':<50}") print(f"{'Clustering':<20} {'KMeans, Bisecting, GMM, LDA':<50}") print(f"{'Recommendation':<20} {'ALS':<50}") print(f"{'Dim reduction':<20} {'PCA, SVD':<50}") print(f"{'Feature':<20} {'Assembler, Scaler, Encoder, TF-IDF':<50}")

🔗 ML Pipelines

🎯 End-to-End ML Workflows

ML Pipelines in MLlib let you chain feature engineering, model training, and evaluation into a single reproducible workflow.

Pipelines

Stages: Transformers + Estimators. Fit: Train all. Transform: Apply. Cross-val: Tune. Save: Persist. Deploy: Production.

🔗 Stages🔗 Fit🔗 Transform🔗 Save
# ML Pipelines print("=== What is a Pipeline? ===") print(""" Pipeline: - Sequence of stages - Transformers - Estimators - Fit + transform - Reproducible Stages: - Transformer: transform() - Estimator: fit() - Pipeline: fit() + transform() Benefits: - Reproducible - Encapsulated - Tunable - Deployable - Testable """) print("\n=== Example Pipeline ===") print(""" from pyspark.ml import Pipeline, PipelineModel from pyspark.ml.feature import ( StringIndexer, OneHotEncoder, VectorAssembler, StandardScaler ) from pyspark.ml.classification import RandomForestClassifier from pyspark.ml.evaluation import BinaryClassificationEvaluator from pyspark.ml.tuning import CrossValidator, ParamGridBuilder # Stages categorical_cols = ["gender", "occupation"] numerical_cols = ["age", "salary", "tenure"] # Encode categorical indexers = [ StringIndexer(inputCol=c, outputCol=f"{c}_idx") for c in categorical_cols ] encoders = [ OneHotEncoder(inputCol=f"{c}_idx", outputCol=f"{c}_vec") for c in categorical_cols ] # Assemble features assembler = VectorAssembler( inputCols=numerical_cols + [f"{c}_vec" for c in categorical_cols], outputCol="features" ) # Scale scaler = StandardScaler( inputCol="features", outputCol="scaled_features" ) # Model rf = RandomForestClassifier( featuresCol="scaled_features", labelCol="label", numTrees=100 ) # Pipeline pipeline = Pipeline(stages=indexers + encoders + [ assembler, scaler, rf ]) # Train model = pipeline.fit(train_df) # Predict predictions = model.transform(test_df) # Evaluate evaluator = BinaryClassificationEvaluator(labelCol="label") auc = evaluator.evaluate(predictions) print(f"AUC: {auc}") # Save model.save("models/rf_pipeline") loaded = PipelineModel.load("models/rf_pipeline") """) print("\n=== Cross-Validation ===") print(""" # Param grid param_grid = ParamGridBuilder() \\ .addGrid(rf.numTrees, [50, 100, 200]) \\ .addGrid(rf.maxDepth, [5, 10, 15]) \\ .build() # Cross-validator cv = CrossValidator( estimator=pipeline, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5, parallelism=4 ) # Fit cv_model = cv.fit(train_df) # Best model best_model = cv_model.bestModel print(f"Best numTrees: {best_model.stages[-1].getNumTrees}") print(f"Best maxDepth: {best_model.stages[-1].getMaxDepth()}") # Average metrics print(f"Avg AUC: {sum(cv_model.avgMetrics) / len(cv_model.avgMetrics)}") """) print("\n=== Train Validation Split ===") print(""" from pyspark.ml.tuning import TrainValidationSplit tvs = TrainValidationSplit( estimator=pipeline, estimatorParamMaps=param_grid, evaluator=evaluator, trainRatio=0.8, parallelism=4 ) tvs_model = tvs.fit(train_df) """) print("\n=== Deployment ===") print(""" # Save model.save("s3://models/rf_pipeline") # Load from pyspark.ml import PipelineModel model = PipelineModel.load("s3://models/rf_pipeline") # Predict streaming stream = spark.readStream \\ .format("kafka") \\ .option("subscribe", "events") \\ .load() predictions = model.transform(stream) predictions.writeStream \\ .format("kafka") \\ .option("topic", "predictions") \\ .start() """) print("\n=== Best Practices ===") print(""" 1. Version your pipeline 2. Save preprocessors 3. Test stages 4. Document 5. Use MLflow 6. Monitor drift 7. Retrain 8. A/B test """)

⚙️ Horovod

🎯 Distributed Deep Learning

Horovod is Uber's open-source distributed deep learning framework. It uses ring-AllReduce for efficient gradient aggregation and supports TensorFlow, PyTorch, and MXNet with minimal code changes.

Horovod

Ring-AllReduce: Efficient. Framework: Multi. Simple: Few lines. Scale: Hundreds of GPUs. Open: Apache 2.0. Production: Uber.

⚙️ Ring-AllReduce⚙️ Multi-Framework⚙️ Simple API⚙️ Production
# Horovod print("=== What is Horovod? ===") print(""" Horovod: - Distributed deep learning - Ring-AllReduce - Multi-framework - Simple API - Production-ready Created by: - Uber (2017) - Open source (Apache 2.0) Supported: - TensorFlow - PyTorch - MXNet - Keras Features: - Ring-AllReduce - NCCL, Gloo, MPI - Gradient compression - Tensor fusion - Timeline - Elastic training """) print("\n=== Why Horovod? ===") print(""" 1. EFFICIENT - Ring-AllReduce - Optimal bandwidth - NCCL for GPU 2. SIMPLE - Few lines of code - Minimal changes - Intuitive API 3. FLEXIBLE - Multiple frameworks - Multiple backends - CPU + GPU 4. PRODUCTION - Uber scale - Battle-tested - Active community """) print("\n=== PyTorch + Horovod ===") print(""" import torch import horovod.torch as hvd # Init hvd.init() # GPU torch.cuda.set_device(hvd.local_rank()) # Model model = MyModel().cuda() # Optimizer optimizer = torch.optim.Adam(model.parameters(), lr=0.001 * hvd.size()) # Wrap with Horovod optimizer = hvd.DistributedOptimizer( optimizer, named_parameters=model.named_parameters() ) # Broadcast initial hvd.broadcast_parameters(model.state_dict(), root_rank=0) hvd.broadcast_optimizer_state(optimizer, root_rank=0) # Data train_sampler = torch.utils.data.distributed.DistributedSampler( dataset, num_replicas=hvd.size(), rank=hvd.rank() ) train_loader = torch.utils.data.DataLoader( dataset, batch_size=32, sampler=train_sampler ) # Train for epoch in range(epochs): train_sampler.set_epoch(epoch) for batch in train_loader: optimizer.zero_grad() output = model(batch) loss = criterion(output, target) loss.backward() # AllReduce optimizer.step() # Save only on rank 0 if hvd.rank() == 0: torch.save(model.state_dict(), "model.pt") """) print("\n=== Run ===") print(""" # Single machine, 4 GPUs horovodrun -np 4 -H localhost:4 python train.py # Multi-machine horovodrun -np 16 -H server1:4,server2:4,server3:4,server4:4 \\ python train.py # With MPI mpirun -np 16 -H server1:4,server2:4,server3:4,server4:4 \\ -bind-to none -map-by slot \\ -x NCCL_DEBUG=INFO \\ -x LD_LIBRARY_PATH \\ -mca pml ob1 -mca btl ^openib \\ python train.py # With Slurm srun --nodes=4 --ntasks-per-node=4 python train.py # With Kubernetes # Use kubeflow mpi-operator or horovod operator """) print("\n=== TensorFlow + Horovod ===") print(""" import tensorflow as tf import horovod.tensorflow.keras as hvd # Init hvd.init() # GPU gpus = tf.config.experimental.list_physical_devices('GPU') for gpu in gpus: tf.config.experimental.set_memory_growth(gpu, True) if gpus: tf.config.experimental.set_visible_devices(gpus[hvd.local_rank()], 'GPU') # Model model = create_model() # Optimizer with scaling opt = tf.keras.optimizers.Adam(0.001 * hvd.size()) opt = hvd.DistributedOptimizer(opt) # Compile model.compile( optimizer=opt, loss='sparse_categorical_crossentropy', metrics=['accuracy'], experimental_run_tf_function=False ) # Callbacks callbacks = [ hvd.callbacks.BroadcastGlobalVariablesCallback(0), hvd.callbacks.MetricAverageCallback(), ] # Train model.fit( train_dataset, epochs=10, callbacks=callbacks, verbose=1 if hvd.rank() == 0 else 0 ) # Save if hvd.rank() == 0: model.save("model.h5") """) print("\n=== Features ===") print(""" 1. RING-ALLREDUCE - Optimal bandwidth - Scales linearly 2. TENSOR FUSION - Combine small tensors - Reduce communication 3. GRADIENT COMPRESSION - Reduce size - Less bandwidth - Trade accuracy 4. TIMELINE - Profile - Debug - Optimize 5. ELASTIC - Add/remove workers - Fault tolerance - Spot instances """) print("\n=== Best Practices ===") print(""" 1. Scale learning rate - lr * num_workers 2. Warmup - First epochs - Stability 3. Batch size - Larger per GPU - Effective batch 4. Checkpointing - Rank 0 only - Broadcast 5. Monitoring - Timeline - Metrics """)

⚖️ MLlib vs Distributed DL

🎯 Two Worlds of Distributed ML

MLlib is for classical ML on tabular data at scale. Distributed deep learning (Horovod, DDP) is for neural networks and large models.

Comparison

MLlib: Classical, tabular. DL: Neural, unstructured. MLlib: Spark. DL: Horovod. MLlib: Simpler. DL: More powerful.

⚖️ MLlib⚖️ DL⚖️ Tabular⚖️ Neural
Aspect MLlib Distributed DL
FocusClassical MLDeep learning
DataTabularUnstructured
FrameworkSparkHorovod, DDP
ParallelismDataData + model
HardwareCPUGPU
ScaleBillions of rowsBillions of params
AlgorithmsLR, RF, GBT, KMeansCNN, RNN, Transformer
PipelinesBuilt-inCustom
ComplexityMediumHigh
Best forTabular, feature engineeringImages, text, LLMs
# When to Use Which print("=== Use MLlib WHEN ===") print(""" ✓ Tabular data ✓ Classical algorithms ✓ Spark ecosystem ✓ Feature engineering ✓ CPU cluster ✓ Billions of rows ✓ Simpler pipelines ✓ Business ML Examples: - Credit scoring - Churn prediction - Recommendations (ALS) - Customer segmentation """) print("\n=== Use Distributed DL WHEN ===") print(""" ✓ Neural networks ✓ Unstructured data ✓ Large models ✓ GPU cluster ✓ Images, text, audio ✓ LLMs ✓ Deep learning ✓ Research Examples: - Image classification - NLP - Transformers - Generative models """) print("\n=== Both ===") print(""" Many pipelines use both: - MLlib: Feature engineering, classical ML - DL: Deep learning on features - MLlib: Aggregation, preprocessing - DL: Model training Example: - Spark: Preprocess 1B rows - MLlib: Feature engineering - Horovod: Train neural net - MLlib: Post-process, evaluate """) print("\n=== Decision Flow ===") print(""" Start │ ├─ Tabular data? ──Yes──→ MLlib │ │ │ No │ ↓ ├─ Neural network? ──Yes──→ Distributed DL │ │ │ No │ ↓ ├─ Spark? ──Yes──→ MLlib │ │ │ No │ ↓ └─ Deep learning? ──Yes──→ Horovod/DDP """)

✅ Best Practices

🎯 Doing Distributed ML Right

Distributed ML is complex. Follow these best practices to succeed.

Best Practices

Start small: Single machine. Profile: Find bottleneck. Scale: Incrementally. Monitor: Metrics. Reproduce: Version. Optimize: Iterate.

✅ Start Small✅ Profile✅ Scale✅ Monitor
# Best Practices print("=== Start Small ===") print(""" 1. SINGLE MACHINE FIRST - Prototype - Validate - Small sample - Iterate 2. PROFILE - Find bottleneck - CPU, GPU, I/O, network - Measure 3. SCALE INCREMENTALLY - 1 → 2 → 4 → 8 - Measure speedup - Linear? 4. OPTIMIZE BOTTLENECK - Communication - I/O - Compute """) print("\n=== Data ===") print(""" 1. PARTITION - Right size - Even distribution - Avoid skew 2. SHUFFLE - Minimize - Colocate - Broadcast small 3. CACHE - Reuse - Memory - Format 4. FORMAT - Parquet - ORC - Columnar """) print("\n=== Model ===") print(""" 1. CHECKPOINT - Regular - Fault tolerance - Resume 2. EARLY STOPPING - Save time - Avoid overfit 3. HYPERPARAMETERS - Scale LR - Batch size - Warmup 4. REGULARIZATION - Avoid overfit - Dropout - Weight decay """) print("\n=== Training ===") print(""" 1. EFFECTIVE BATCH SIZE - Global = local × workers - Scale LR - Warmup 2. SYNCHRONIZATION - Every step - Bucket - Overlap 3. GRADIENT ACCUMULATION - Simulate larger batch - Memory efficient 4. MIXED PRECISION - FP16/BF16 - 2x speed - Memory savings """) print("\n=== Operations ===") print(""" 1. MONITOR - Loss - Throughput - GPU utilization - Communication 2. LOG - Metrics - Config - Versions 3. REPRODUCE - Seeds - Versions - Config 4. VERSION - Data - Model - Code """) print("\n=== Anti-Patterns ===") print(""" AVOID: ✗ Big bang scaling ✗ Ignoring communication ✗ Data skew ✗ Small batches ✗ Wrong LR ✗ No checkpointing ✗ No monitoring ✗ No versioning """) print("\n=== Checklist ===") print(""" ✓ Start single machine ✓ Profile ✓ Scale incrementally ✓ Right batch size ✓ Scale LR ✓ Warmup ✓ Checkpoint ✓ Monitor ✓ Reproduce ✓ Version """)

❓ FAQ

🎯 Frequently Asked Questions

Answers to common questions about MLlib and distributed ML.

What is MLlib?

MLlib is Apache Spark's scalable machine learning library. It provides distributed implementations of common ML algorithms (classification, regression, clustering, collaborative filtering) that run on Spark clusters, scaling to billions of records.

What is distributed machine learning?

Distributed machine learning trains models across multiple machines to handle datasets too large for a single machine. Common approaches include data parallelism (splitting data), model parallelism (splitting model), and hybrid approaches using parameter servers or AllReduce.

What is the difference between data parallelism and model parallelism?

Data parallelism splits the training data across workers, each training a copy of the model, then aggregating gradients. Model parallelism splits the model itself across workers, each computing a portion of the model. Data parallelism is more common and simpler; model parallelism is used when models are too large for a single GPU.

What is Horovod?

Horovod is an open-source distributed deep learning framework from Uber. It uses ring-AllReduce for efficient gradient aggregation across workers, supports TensorFlow, PyTorch, and MXNet, and scales to hundreds of GPUs with minimal code changes.

When should I use MLlib vs distributed deep learning?

Use MLlib for classical ML algorithms (logistic regression, random forests, K-means, ALS) on tabular data at scale. Use distributed deep learning (Horovod, PyTorch DDP, TensorFlow Distributed) for neural networks, deep learning, and large models like transformers or CNNs.

What is AllReduce?

AllReduce is a collective communication operation that aggregates values across all workers and distributes the result to all. It's the foundation of modern distributed training (Horovod, NCCL). Ring-AllReduce is bandwidth-optimal.

What is a parameter server?

A parameter server is a centralized architecture where a server (or group of servers) stores model parameters. Workers compute gradients and push them to the server, which aggregates and updates. It's an alternative to AllReduce, sometimes better for very large models or asynchronous training.

How do I scale my learning rate for distributed training?

When you multiply the batch size by N workers, scale the learning rate by N (linear scaling) or √N (square root scaling). Use a warmup period for the first few epochs to stabilize training. Modern optimizers like LAMB and LARS handle this automatically.

What is the difference between synchronous and asynchronous training?

Synchronous training waits for all workers to finish each step before aggregating gradients (deterministic, simpler). Asynchronous training lets workers update independently (faster but uses stale gradients, less stable). Most modern frameworks default to synchronous.

Can I use MLlib with deep learning?

Yes. MLlib's MultilayerPerceptronClassifier supports simple neural networks. For deep learning at scale, use Spark for preprocessing and feature engineering, then Horovod or PyTorch DDP for training. You can also use SparkTorch or DeepLearning4J on Spark.

# Quick Reference print("=== Quick Answers ===") print(""" Q: MLlib for what? A: Classical ML on tabular data at scale. Q: Horovod for what? A: Distributed deep learning. Q: Data or model parallel? A: Data parallel most common. Q: Sync or async? A: Sync (simpler, deterministic). Q: How to scale LR? A: lr * num_workers + warmup. Q: Best framework for GPU? A: PyTorch DDP + NCCL. Q: Best for tabular? A: Spark MLlib. Q: Best for LLMs? A: PyTorch + Horovod/Megatron. """)

🚀 Future

🎯 Where Distributed ML Is Going

The future is larger models, more efficient training, auto-parallelism, and AI-driven optimization. Distributed ML is the foundation of modern AI.

Future Directions

Scale: Trillion params. Efficiency: Better algorithms. Auto: Parallelism. AI: Optimization. Hardware: Specialized. Democratize: Accessible.

🚀 Scale🚀 Efficiency🚀 Auto🚀 AI
# Future of Distributed ML print("=== 1. Larger Models ===") print(""" Trend: Trillion-parameter models Current: - GPT-4: ~1.7T params - PaLM: 540B params - Llama 3: 405B params Future: - 10T+ params - Multi-modal - MoE (Mixture of Experts) - Sparse models Challenges: - Memory - Communication - Cost - Energy """) print("\n=== 2. Efficiency ===") print(""" Trend: Train faster, cheaper Techniques: - Mixed precision (FP8) - Gradient compression - ZeRO (DeepSpeed) - FlashAttention - LoRA / QLoRA - Quantization Future: - 10x faster - 10x cheaper - 10x accessible """) print("\n=== 3. Auto-Parallelism ===") print(""" Trend: Automatic distribution Auto: - Detect parallelism - Choose strategy - Optimize - Schedule Tools: - JAX (pjit, pmap) - PyTorch FSDP - DeepSpeed - Alpa - FlexFlow Future: - No manual - Optimal - Portable """) print("\n=== 4. AI for ML ===") print(""" Trend: AI optimizes training AI: - Hyperparameter tuning - Architecture search - Parallelism choice - Resource allocation Examples: - Google Vizier - Ray Tune - Optuna - AutoML Future: - Self-optimizing - Self-healing - Autonomous """) print("\n=== 5. Specialized Hardware ===") print(""" Trend: Purpose-built chips Hardware: - Google TPU - NVIDIA H100/B200 - AMD MI300 - AWS Trainium - Graphcore - Cerebras Features: - Higher bandwidth - Better interconnect - Efficient Future: - More specialized - Faster - Cheaper """) print("\n=== 6. Democratization ===") print(""" Trend: Accessible to all Democratize: - Small teams - Open models - Cloud services - Fine-tuning - LoRA Examples: - Hugging Face - Together AI - Anyscale - Modal Future: - Anyone can train - Open models - Accessible """) # TIMELINE print("\n=== Timeline ===") print(""" Past: 2013: MLlib 2017: Horovod 2018: PyTorch DDP 2020: ZeRO, DeepSpeed 2022: Megatron-LM 2023: LoRA, QLoRA Present: 2024: Large models 2025: Auto-parallelism Future: 2030: Auto everything? 2035: New paradigms? 2040: What's next? """) print("\n=== Final Thoughts ===") print(""" Distributed ML is: - Foundation of AI - Scaling to billions - Complex but powerful - Evolving fast The future is: - Larger models - More efficient - Auto-parallel - AI-driven - Accessible "Scale is the new frontier. Distribute to innovate." Keep exploring. Keep building. """)