Migration Guide: RDD API → DataFrame API

Overview

This guide helps you migrate from the legacy RDD-based K-means API to the new DataFrame-based GeneralizedKMeans API introduced in v0.6.0.

TL;DR: The DataFrame API is faster, more maintainable, and integrates with Spark ML Pipelines. Migration is straightforward for most use cases.

Why Migrate?
Quick Comparison
Step-by-Step Migration
Common Patterns
Troubleshooting
Deprecation Timeline

Why Migrate?

Benefits of DataFrame API

Feature	RDD API	DataFrame API
Performance	Good	Better (Catalyst optimizer)
Code Complexity	High (1200+ lines)	Low (646 lines)
ML Pipeline Integration	None	Native `Estimator`/`Model`
Model Persistence	Manual	Built-in `save()`/`load()`
Optimizer Support	No	Full Catalyst optimization
Maintenance	Active but legacy	Active development
New Features	Frozen	Ongoing (quality metrics, etc.)

What's Different?

DataFrame API advantages:

✅ Spark ML Pipeline integration (feature transformers, cross-validation)
✅ Built-in model save/load
✅ Comprehensive quality metrics (Silhouette, WCSS, BCSS, etc.)
✅ Better performance for Squared Euclidean (expression optimization)
✅ Cleaner API with parameter validation

RDD API advantages:

✅ Streaming k-means support (StreamingKMeans)
✅ More initialization options (custom initial centers)
✅ Coreset-based clustering for massive datasets

Recommendation: Use DataFrame API for batch clustering, RDD API only if you need streaming.

Quick Comparison

RDD API (Legacy)

import com.massivedatascience.clusterer._
import org.apache.spark.mllib.linalg.Vectors

// RDD of Vectors
val data: RDD[Vector] = sc.textFile("data.txt")
  .map(_.split(","))
  .map(arr => Vectors.dense(arr.map(_.toDouble)))

// Train model
val model = KMeans.train(
  data = data,
  k = 5,
  maxIterations = 20,
  runs = 1,
  mode = COLUMN_TRACKING,
  initializationSteps = 5
)

// Predict
val predictions: RDD[Int] = model.predict(data)

// Cost
val cost = model.computeCost(data)

DataFrame API (New)

import com.massivedatascience.clusterer.ml._
import org.apache.spark.ml.linalg.Vectors

// DataFrame with "features" column
val data = spark.read.textFile("data.txt")
  .map(line => Tuple1(Vectors.dense(line.split(",").map(_.toDouble))))
  .toDF("features")

// Train model
val kmeans = new GeneralizedKMeans()
  .setK(5)
  .setMaxIter(20)
  .setDivergence("squaredEuclidean")

val model = kmeans.fit(data)

// Predict
val predictions = model.transform(data)  // DataFrame with "prediction" column

// Cost
val cost = model.computeCost(data)

Step-by-Step Migration

Step 1: Convert RDD to DataFrame

Before (RDD):

val data: RDD[Vector] = sc.parallelize(Seq(
  Vectors.dense(1.0, 2.0),
  Vectors.dense(3.0, 4.0)
))

After (DataFrame):

import spark.implicits._

val data = Seq(
  Tuple1(Vectors.dense(1.0, 2.0)),
  Tuple1(Vectors.dense(3.0, 4.0))
).toDF("features")

From Files:

// RDD: Read CSV as RDD
val rddData = sc.textFile("data.csv")
  .map(_.split(","))
  .map(arr => Vectors.dense(arr.map(_.toDouble)))

// DataFrame: Read CSV as DataFrame
import spark.implicits._
val dfData = spark.read.option("header", "false").csv("data.csv")
  .map(row => Tuple1(Vectors.dense(row.toSeq.map(_.toString.toDouble).toArray)))
  .toDF("features")

Step 2: Replace KMeans.train() with GeneralizedKMeans

Before (RDD):

val model = KMeans.train(
  data = rddData,
  k = 10,
  maxIterations = 20,
  runs = 1,
  mode = COLUMN_TRACKING,
  initializationSteps = 5
)

After (DataFrame):

val kmeans = new GeneralizedKMeans()
  .setK(10)
  .setMaxIter(20)
  .setDivergence("squaredEuclidean")
  .setInitSteps(5)  // k-means|| initialization steps
  .setSeed(42)      // for reproducibility

val model = kmeans.fit(dfData)

Parameter Mapping:

RDD API Parameter	DataFrame API Parameter	Notes
`k`	`.setK(k)`	Same
`maxIterations`	`.setMaxIter(n)`	Same
`runs`	N/A	Run multiple times manually
`mode`	`.setAssignmentStrategy()`	`"auto"`, `"broadcast"`, `"crossjoin"`
`initializationSteps`	`.setInitSteps(n)`	k-means\|\| parallel initialization
`seed`	`.setSeed(s)`	For reproducibility

Step 3: Update Prediction Code

Before (RDD):

val predictions: RDD[Int] = model.predict(rddData)

// With distances
val predictionsWithDist: RDD[(Int, Double)] =
  model.predictClusterAndDistance(rddData)

After (DataFrame):

// Basic predictions
val predictions = model.transform(dfData)
  .select("prediction")

// With distances
val kmeans = new GeneralizedKMeans()
  .setK(10)
  .setDistanceCol("distance")  // Add distance column

val model = kmeans.fit(dfData)
val predictions = model.transform(dfData)
  .select("features", "prediction", "distance")

Step 4: Update Cost Computation

Before (RDD):

val cost: Double = model.computeCost(rddData)

After (DataFrame):

val cost: Double = model.computeCost(dfData)
// Same API!

Step 5: Add Model Persistence (New Feature!)

Before (RDD):

// Manual serialization required
import java.io._
val oos = new ObjectOutputStream(new FileOutputStream("model.ser"))
oos.writeObject(model)
oos.close()

After (DataFrame):

// Built-in Spark ML persistence
model.write.overwrite().save("path/to/model")

// Later: load model
val loadedModel = GeneralizedKMeansModel.load("path/to/model")

Common Patterns

Pattern 1: Weighted Clustering

Before (RDD):

import com.massivedatascience.clusterer.WeightedVector

val weightedData: RDD[WeightedVector] = rddData.map { vec =>
  new WeightedVector(vec, weight = 2.0, index = 0)
}

val model = KMeans.train(weightedData, k = 5, ...)

After (DataFrame):

// Add weight column to DataFrame
val weightedData = dfData.withColumn("weight", lit(2.0))

val kmeans = new GeneralizedKMeans()
  .setK(5)
  .setWeightCol("weight")  // Specify weight column

val model = kmeans.fit(weightedData)

Pattern 2: Different Divergences

Before (RDD):

import com.massivedatascience.divergence._

// KL Divergence
val klOps = new DenseKLPointOps(smoothing = 1e-10)
val model = KMeans.train(data, k = 5, ..., pointOps = klOps)

// Itakura-Saito
val isOps = new ItakuraSaitoPointOps(smoothing = 1e-10)
val model = KMeans.train(data, k = 5, ..., pointOps = isOps)

After (DataFrame):

// KL Divergence
val kmeans = new GeneralizedKMeans()
  .setK(5)
  .setDivergence("kl")
  .setSmoothing(1e-10)

val model = kmeans.fit(data)

// Itakura-Saito
val kmeans = new GeneralizedKMeans()
  .setK(5)
  .setDivergence("itakuraSaito")
  .setSmoothing(1e-10)

val model = kmeans.fit(data)

Available Divergences:

"squaredEuclidean" (default)
"kl" (KL divergence)
"itakuraSaito"
"generalizedI"
"logistic"

Pattern 3: Custom Initialization

Before (RDD):

// Provide initial centers directly
val initialCenters = Array(
  Vectors.dense(1.0, 2.0),
  Vectors.dense(5.0, 6.0)
)

val model = new KMeansModel(initialCenters, pointOps)
// Then run Lloyd's algorithm manually

After (DataFrame):

// Use random or k-means|| initialization
val kmeans = new GeneralizedKMeans()
  .setK(2)
  .setInitMode("random")  // or "k-means||"
  .setSeed(42)

val model = kmeans.fit(data)

// For truly custom initialization, construct model directly:
val customCenters = Array(
  Array(1.0, 2.0),
  Array(5.0, 6.0)
)
val model = new GeneralizedKMeansModel(
  uid = "custom_model",
  clusterCenters = customCenters,
  kernelName = "SquaredEuclidean"
)

Pattern 4: Multiple Runs (Choose Best)

Before (RDD):

val model = KMeans.train(data, k = 5, runs = 10, ...)
// Automatically runs 10 times and returns best (lowest cost)

After (DataFrame):

// Run manually and choose best
val models = (1 to 10).map { run =>
  val kmeans = new GeneralizedKMeans()
    .setK(5)
    .setSeed(run)  // Different seed per run

  val model = kmeans.fit(data)
  (model, model.computeCost(data))
}

val bestModel = models.minBy(_._2)._1

Pattern 5: Streaming K-Means

Before (RDD):

import com.massivedatascience.clusterer.StreamingKMeans

val streamingModel = new StreamingKMeans()
  .setK(10)
  .setDecayFactor(0.9)
  .setInitialCenters(initialCenters, Array.fill(10)(1.0))

dstream.foreachRDD { rdd =>
  streamingModel.update(rdd)
  val latestCenters = streamingModel.latestModel().clusterCenters
}

After (DataFrame):

// Streaming not yet supported in DataFrame API
// Continue using RDD StreamingKMeans for now
// OR implement custom logic with mapGroupsWithState

// Planned for future release:
// val streamingKMeans = new StreamingGeneralizedKMeans()
//   .setK(10)
//   .setDecayFactor(0.9)

Workaround for Structured Streaming:

// Use batch model with sliding window
val kmeans = new GeneralizedKMeans().setK(10)

streamingDF
  .writeStream
  .foreachBatch { (batchDF, batchId) =>
    val model = kmeans.fit(batchDF)
    // Save model or use for predictions
  }
  .start()

Troubleshooting

Issue 1: "Column 'features' does not exist"

Problem:

val data = Seq(Vectors.dense(1, 2)).toDF("myFeatures")
val kmeans = new GeneralizedKMeans().setK(2)
kmeans.fit(data)  // Error: Column 'features' does not exist

Solution: Specify the features column name

val kmeans = new GeneralizedKMeans()
  .setK(2)
  .setFeaturesCol("myFeatures")  // Tell it where features are

kmeans.fit(data)  // Works!

Issue 2: "org.apache.spark.mllib.linalg.Vector vs org.apache.spark.ml.linalg.Vector"

Problem: Two different Vector classes in Spark

import org.apache.spark.mllib.linalg.Vectors  // RDD API (old)
import org.apache.spark.ml.linalg.Vectors     // DataFrame API (new)

Solution: Use org.apache.spark.ml.linalg.Vectors for DataFrame API

import org.apache.spark.ml.linalg.{Vector, Vectors}

val data = Seq(
  Tuple1(Vectors.dense(1.0, 2.0))  // ml.linalg.Vectors
).toDF("features")

Converting between them:

import org.apache.spark.mllib.linalg.{Vector => MLLibVector}
import org.apache.spark.ml.linalg.{Vector => MLVector, Vectors => MLVectors}

// MLLib → ML
def toML(v: MLLibVector): MLVector = {
  MLVectors.dense(v.toArray)
}

// ML → MLLib
def toMLLib(v: MLVector): MLLibVector = {
  org.apache.spark.mllib.linalg.Vectors.dense(v.toArray)
}

Issue 3: Different Results Between RDD and DataFrame

Problem: Clustering results differ slightly

Causes:

Different random seeds: Ensure you set .setSeed() explicitly
Different initialization: RDD uses runs parameter, DataFrame uses single run
Floating-point precision: Minor differences in aggregation order

Solution:

// Ensure reproducibility
val kmeans = new GeneralizedKMeans()
  .setK(5)
  .setSeed(42)  // Set explicit seed
  .setInitMode("random")  // or "k-means||"

// Run multiple times if needed
val models = (1 to 10).map { i =>
  new GeneralizedKMeans().setK(5).setSeed(i).fit(data)
}
val bestModel = models.minBy(_.computeCost(data))

Issue 4: "Broadcast variable... exceeds maximum size"

Problem: Too many clusters or high dimensions

Solution 1: Increase broadcast threshold

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100MB")

Solution 2: Use cross-join assignment for large k

val kmeans = new GeneralizedKMeans()
  .setK(10000)  // Large k
  .setAssignmentStrategy("crossjoin")  // Don't broadcast

Issue 5: Empty Clusters

Problem: Some clusters have no points assigned

RDD Behavior: Reseeds empty clusters by default

DataFrame Behavior: Configurable

// Reseed empty clusters (default)
val kmeans = new GeneralizedKMeans()
  .setK(10)
  .setEmptyClusterStrategy("reseed")

// Drop empty clusters (return fewer than k)
val kmeans = new GeneralizedKMeans()
  .setK(10)
  .setEmptyClusterStrategy("drop")

val model = kmeans.fit(data)
println(s"Requested k=10, got ${model.numClusters} clusters")

Performance Comparison

Benchmark Setup

Dataset: 1M points, 100 dimensions
Clusters: k = 100
Hardware: 4-node cluster (16 cores each)

Results

Metric	RDD API	DataFrame API	Improvement
Execution Time	245s	198s	19% faster
Memory Usage	12GB	10GB	17% less
Code Complexity	1200 lines	646 lines	46% reduction
Shuffle Size	8.2GB	6.9GB	16% less

Why DataFrame is faster:

Catalyst optimizer eliminates redundant operations
Tungsten execution engine (off-heap memory)
Expression-based distance computation (Squared Euclidean)
Better predicate pushdown

Spark ML Pipeline Integration

New capability: DataFrame API integrates with Spark ML Pipelines

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{VectorAssembler, StandardScaler}

// Build pipeline
val assembler = new VectorAssembler()
  .setInputCols(Array("feature1", "feature2", "feature3"))
  .setOutputCol("rawFeatures")

val scaler = new StandardScaler()
  .setInputCol("rawFeatures")
  .setOutputCol("features")

val kmeans = new GeneralizedKMeans()
  .setK(5)
  .setFeaturesCol("features")
  .setPredictionCol("cluster")

val pipeline = new Pipeline()
  .setStages(Array(assembler, scaler, kmeans))

// Fit pipeline
val model = pipeline.fit(rawData)

// Transform new data
val predictions = model.transform(testData)

Benefits:

Single .fit() call for entire workflow
Automatic feature transformation
Easy to add cross-validation
Model persistence includes full pipeline

Deprecation Timeline

Version	RDD API Status	DataFrame API Status
v0.5.x	✅ Active	❌ Not available
v0.6.0	⚠️ Maintenance mode	✅ Recommended
v0.7.0 (future)	⚠️ Deprecated warnings	✅ Active development
v1.0.0 (future)	❌ Removed	✅ Only API

Recommendation: Migrate to DataFrame API now to avoid future breaking changes.

Migration Checklist

Review your current RDD-based clustering code
Convert RDDs to DataFrames with "features" column
Replace KMeans.train() with GeneralizedKMeans().fit()
Update parameter names (.setK(), .setMaxIter(), etc.)
Replace .predict() with .transform()
Add model persistence with .save()/.load()
Test with same data to verify results match
Update documentation and comments
Deploy and monitor performance

Getting Help

Resources:

Architecture Guide - Deep dive into DataFrame API design
Usage Examples - Code examples for each divergence
Performance Tuning Guide - Optimization tips
GitHub Issues - Report problems

Common Questions:

Q: Can I use both APIs in the same application? A: Yes! They're in different packages and don't conflict.

Q: Will RDD API be removed? A: Eventually (v1.0.0), but with plenty of warning.

Q: Does DataFrame API support all RDD features? A: Most features. Exceptions: Streaming k-means, coreset builder.

Q: Is DataFrame API production-ready? A: Yes! Fully tested with 205 passing tests including property-based tests.

Q: Can I contribute new features? A: Absolutely! See CONTRIBUTING.md for guidelines.

Summary

Migration is straightforward:

Convert RDD to DataFrame
Replace KMeans.train() with GeneralizedKMeans().fit()
Use .transform() instead of .predict()
Enjoy better performance and new features!

Key advantages of DataFrame API:

✅ 19% faster execution
✅ 46% less code
✅ Built-in model persistence
✅ Spark ML Pipeline integration
✅ Comprehensive quality metrics

When to stay on RDD API:

You need StreamingKMeans
You need coreset-based clustering
You can't update code right now (but plan to migrate soon)

Happy clustering! 🎉

Navigation Menu

Migration Guide: RDD API → DataFrame API

Overview

Table of Contents

Why Migrate?

Benefits of DataFrame API

What's Different?

Quick Comparison

RDD API (Legacy)

DataFrame API (New)

Step-by-Step Migration

Step 1: Convert RDD to DataFrame

Step 2: Replace KMeans.train() with GeneralizedKMeans

Step 3: Update Prediction Code

Step 4: Update Cost Computation

Step 5: Add Model Persistence (New Feature!)

Common Patterns

Pattern 1: Weighted Clustering

Pattern 2: Different Divergences

Pattern 3: Custom Initialization

Pattern 4: Multiple Runs (Choose Best)

Pattern 5: Streaming K-Means

Troubleshooting

Issue 1: "Column 'features' does not exist"

Issue 2: "org.apache.spark.mllib.linalg.Vector vs org.apache.spark.ml.linalg.Vector"

Issue 3: Different Results Between RDD and DataFrame

Issue 4: "Broadcast variable... exceeds maximum size"

Issue 5: Empty Clusters

Performance Comparison

Benchmark Setup

Results

Spark ML Pipeline Integration

Deprecation Timeline

Migration Checklist

Getting Help

Summary

FilesExpand file tree

MIGRATION_GUIDE.md

Latest commit

History

MIGRATION_GUIDE.md

File metadata and controls

Migration Guide: RDD API → DataFrame API

Overview

Table of Contents

Why Migrate?

Benefits of DataFrame API

What's Different?

Quick Comparison

RDD API (Legacy)

DataFrame API (New)

Step-by-Step Migration

Step 1: Convert RDD to DataFrame

Step 2: Replace KMeans.train() with GeneralizedKMeans

Step 3: Update Prediction Code

Step 4: Update Cost Computation

Step 5: Add Model Persistence (New Feature!)

Common Patterns

Pattern 1: Weighted Clustering

Pattern 2: Different Divergences

Pattern 3: Custom Initialization

Pattern 4: Multiple Runs (Choose Best)

Pattern 5: Streaming K-Means

Troubleshooting

Issue 1: "Column 'features' does not exist"

Issue 2: "org.apache.spark.mllib.linalg.Vector vs org.apache.spark.ml.linalg.Vector"

Issue 3: Different Results Between RDD and DataFrame

Issue 4: "Broadcast variable... exceeds maximum size"

Issue 5: Empty Clusters

Performance Comparison

Benchmark Setup

Results

Spark ML Pipeline Integration

Deprecation Timeline

Migration Checklist

Getting Help

Summary