Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-8519][SPARK-11560] [ML] [MLlib] Optimize KMeans implementation. #14937

Closed
wants to merge 6 commits into from
Closed

[SPARK-8519][SPARK-11560] [ML] [MLlib] Optimize KMeans implementation. #14937

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
ce36c51
Optimize KMeans implementation.
yanboliang Sep 2, 2016
e10e630
Swap arguments order of MLUtils.fastSquaredDistance.
yanboliang Sep 3, 2016
9b1858e
Optimization for SparseMatrix.
yanboliang Sep 4, 2016
5bcd70e
Optimize initKMeansParallel.
yanboliang Sep 5, 2016
34b4cdf
Deprecate runs.
yanboliang Sep 5, 2016
1c31cda
Fix SparkR kmeans test.
yanboliang Sep 5, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion R/pkg/inst/tests/testthat/test_mllib.R
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -323,7 +323,7 @@ test_that("spark.kmeans", {
model <- spark.kmeans(data = training, ~ ., k = 2, maxIter = 10, initMode = "random")
sample <- take(select(predict(model, training), "prediction"), 1)
expect_equal(typeof(sample$prediction), "integer")
expect_equal(sample$prediction, 1)
expect_equal(sample$prediction, 0)

# Test stats::kmeans is working
statsModel <- kmeans(x = newIris, centers = 2)
Expand Down
27 changes: 26 additions & 1 deletion mllib/src/main/scala/org/apache/spark/ml/clustering/KMeans.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,25 @@ private[clustering] trait KMeansParams extends Params with HasMaxIter with HasFe
@Since("1.5.0")
def getInitSteps: Int = $(initSteps)

/**
* Block size for stacking input data in matrices to speed up the computation.
* Data is stacked within partitions. If block size is more than remaining data in
* a partition then it is adjusted to the size of this data.
* Recommended size is between 4096 and 8192.
* Default: 4096
* @group expertParam
*/
@Since("2.1.0")
final val blockSize: IntParam = new IntParam(this, "blockSize",
"Block size for stacking input data in matrices. Data is stacked within partitions." +
" If block size is more than remaining data in a partition then " +
"it is adjusted to the size of this data. Recommended size is between 4096 and 8192",
ParamValidators.gt(0))

/** @group expertGetParam */
@Since("2.1.0")
final def getBlockSize: Int = $(blockSize)

/**
* Validates and transforms the input schema.
* @param schema input schema
Expand Down Expand Up @@ -263,7 +282,8 @@ class KMeans @Since("1.5.0") (
maxIter -> 20,
initMode -> MLlibKMeans.K_MEANS_PARALLEL,
initSteps -> 5,
tol -> 1e-4)
tol -> 1e-4,
blockSize -> 4096)

@Since("1.5.0")
override def copy(extra: ParamMap): KMeans = defaultCopy(extra)
Expand Down Expand Up @@ -291,6 +311,10 @@ class KMeans @Since("1.5.0") (
@Since("1.5.0")
def setInitSteps(value: Int): this.type = set(initSteps, value)

/** @group expertSetParam */
@Since("2.1.0")
def setBlockSize(value: Int): this.type = set(blockSize, value)

/** @group setParam */
@Since("1.5.0")
def setMaxIter(value: Int): this.type = set(maxIter, value)
Expand Down Expand Up @@ -320,6 +344,7 @@ class KMeans @Since("1.5.0") (
.setMaxIterations($(maxIter))
.setSeed($(seed))
.setEpsilon($(tol))
.setBlockSize($(blockSize))
val parentModel = algo.run(rdd, Option(instr))
val model = copyValues(new KMeansModel(uid, parentModel).setParent(this))
val summary = new KMeansSummary(
Expand Down
Loading