This is an automated email from the ASF dual-hosted git repository. zero323 pushed a commit to branch branch-3.3 in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.3 by this push: new f4457e6310f [SPARK-37402][PYTHON][MLLIB] Inline typehints for pyspark.mllib.clustering f4457e6310f is described below commit f4457e6310f6bd900d7634b279606436a6faf8fb Author: zero323 <mszymkiew...@gmail.com> AuthorDate: Mon Apr 11 11:29:47 2022 +0200 [SPARK-37402][PYTHON][MLLIB] Inline typehints for pyspark.mllib.clustering ### What changes were proposed in this pull request? This PR migrates type `pyspark.mllib.clustering` annotations from stub file to inline type hints. ### Why are the changes needed? Part of ongoing migration of type hints. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests. Closes #35578 from zero323/SPARK-37402. Authored-by: zero323 <mszymkiew...@gmail.com> Signed-off-by: zero323 <mszymkiew...@gmail.com> (cherry picked from commit e71cf3907b9ff2036dfe45bc8fe939f20cca741b) Signed-off-by: zero323 <mszymkiew...@gmail.com> --- python/pyspark/mllib/clustering.py | 247 ++++++++++++++++++++++++------------ python/pyspark/mllib/clustering.pyi | 188 --------------------------- 2 files changed, 163 insertions(+), 272 deletions(-) diff --git a/python/pyspark/mllib/clustering.py b/python/pyspark/mllib/clustering.py index a9e4fd82089..fd33887fd9e 100644 --- a/python/pyspark/mllib/clustering.py +++ b/python/pyspark/mllib/clustering.py @@ -19,7 +19,9 @@ import sys import array as pyarray from math import exp, log from collections import namedtuple +from typing import Any, List, Optional, Tuple, TypeVar, Union, overload, TYPE_CHECKING +import numpy as np from numpy import array, random, tile from pyspark import SparkContext, since @@ -30,6 +32,12 @@ from pyspark.mllib.stat.distribution import MultivariateGaussian from pyspark.mllib.util import Saveable, Loader, inherit_doc, JavaLoader, JavaSaveable from pyspark.streaming import DStream +if TYPE_CHECKING: + from py4j.java_gateway import JavaObject + from pyspark.mllib._typing import VectorLike + +T = TypeVar("T") + __all__ = [ "BisectingKMeansModel", "BisectingKMeans", @@ -67,24 +75,32 @@ class BisectingKMeansModel(JavaModelWrapper): 0.0 """ - def __init__(self, java_model): + def __init__(self, java_model: "JavaObject"): super(BisectingKMeansModel, self).__init__(java_model) self.centers = [c.toArray() for c in self.call("clusterCenters")] - @property + @property # type: ignore[misc] @since("2.0.0") - def clusterCenters(self): + def clusterCenters(self) -> List[np.ndarray]: """Get the cluster centers, represented as a list of NumPy arrays.""" return self.centers - @property + @property # type: ignore[misc] @since("2.0.0") - def k(self): + def k(self) -> int: """Get the number of clusters""" return self.call("k") - def predict(self, x): + @overload + def predict(self, x: "VectorLike") -> int: + ... + + @overload + def predict(self, x: RDD["VectorLike"]) -> RDD[int]: + ... + + def predict(self, x: Union["VectorLike", RDD["VectorLike"]]) -> Union[int, RDD[int]]: """ Find the cluster that each of the points belongs to in this model. @@ -111,7 +127,7 @@ class BisectingKMeansModel(JavaModelWrapper): x = _convert_to_vector(x) return self.call("predict", x) - def computeCost(self, x): + def computeCost(self, x: Union["VectorLike", RDD["VectorLike"]]) -> float: """ Return the Bisecting K-means cost (sum of squared distances of points to their nearest center) for this model on the given @@ -159,7 +175,14 @@ class BisectingKMeans: """ @classmethod - def train(self, rdd, k=4, maxIterations=20, minDivisibleClusterSize=1.0, seed=-1888008604): + def train( + self, + rdd: RDD["VectorLike"], + k: int = 4, + maxIterations: int = 20, + minDivisibleClusterSize: float = 1.0, + seed: int = -1888008604, + ) -> BisectingKMeansModel: """ Runs the bisecting k-means algorithm return the model. @@ -197,7 +220,7 @@ class BisectingKMeans: @inherit_doc -class KMeansModel(Saveable, Loader): +class KMeansModel(Saveable, Loader["KMeansModel"]): """A clustering model derived from the k-means method. @@ -255,22 +278,30 @@ class KMeansModel(Saveable, Loader): [array([-1000., -1000.]), array([ 5., 5.]), array([ 1000., 1000.])] """ - def __init__(self, centers): + def __init__(self, centers: List["VectorLike"]): self.centers = centers - @property + @property # type: ignore[misc] @since("1.0.0") - def clusterCenters(self): + def clusterCenters(self) -> List["VectorLike"]: """Get the cluster centers, represented as a list of NumPy arrays.""" return self.centers - @property + @property # type: ignore[misc] @since("1.4.0") - def k(self): + def k(self) -> int: """Total number of clusters.""" return len(self.centers) - def predict(self, x): + @overload + def predict(self, x: "VectorLike") -> int: + ... + + @overload + def predict(self, x: RDD["VectorLike"]) -> RDD[int]: + ... + + def predict(self, x: Union["VectorLike", RDD["VectorLike"]]) -> Union[int, RDD[int]]: """ Find the cluster that each of the points belongs to in this model. @@ -297,13 +328,13 @@ class KMeansModel(Saveable, Loader): x = _convert_to_vector(x) for i in range(len(self.centers)): - distance = x.squared_distance(self.centers[i]) + distance = x.squared_distance(self.centers[i]) # type: ignore[attr-defined] if distance < best_distance: best = i best_distance = distance return best - def computeCost(self, rdd): + def computeCost(self, rdd: RDD["VectorLike"]) -> float: """ Return the K-means cost (sum of squared distances of points to their nearest center) for this model on the given @@ -324,20 +355,24 @@ class KMeansModel(Saveable, Loader): return cost @since("1.4.0") - def save(self, sc, path): + def save(self, sc: SparkContext, path: str) -> None: """ Save this model to the given path. """ + assert sc._jvm is not None + java_centers = _py2java(sc, [_convert_to_vector(c) for c in self.centers]) java_model = sc._jvm.org.apache.spark.mllib.clustering.KMeansModel(java_centers) java_model.save(sc._jsc.sc(), path) @classmethod @since("1.4.0") - def load(cls, sc, path): + def load(cls, sc: SparkContext, path: str) -> "KMeansModel": """ Load a model from the given path. """ + assert sc._jvm is not None + java_model = sc._jvm.org.apache.spark.mllib.clustering.KMeansModel.load(sc._jsc.sc(), path) return KMeansModel(_java2py(sc, java_model.clusterCenters())) @@ -352,16 +387,16 @@ class KMeans: @classmethod def train( cls, - rdd, - k, - maxIterations=100, - initializationMode="k-means||", - seed=None, - initializationSteps=2, - epsilon=1e-4, - initialModel=None, - distanceMeasure="euclidean", - ): + rdd: RDD["VectorLike"], + k: int, + maxIterations: int = 100, + initializationMode: str = "k-means||", + seed: Optional[int] = None, + initializationSteps: int = 2, + epsilon: float = 1e-4, + initialModel: Optional[KMeansModel] = None, + distanceMeasure: str = "euclidean", + ) -> "KMeansModel": """ Train a k-means clustering model. @@ -428,7 +463,7 @@ class KMeans: @inherit_doc -class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader): +class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader["GaussianMixtureModel"]): """ A clustering model derived from the Gaussian Mixture Model method. @@ -497,18 +532,18 @@ class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader): True """ - @property + @property # type: ignore[misc] @since("1.4.0") - def weights(self): + def weights(self) -> np.ndarray: """ Weights for each Gaussian distribution in the mixture, where weights[i] is the weight for Gaussian i, and weights.sum == 1. """ return array(self.call("weights")) - @property + @property # type: ignore[misc] @since("1.4.0") - def gaussians(self): + def gaussians(self) -> List[MultivariateGaussian]: """ Array of MultivariateGaussian where gaussians[i] represents the Multivariate Gaussian (Normal) Distribution for Gaussian i. @@ -517,13 +552,21 @@ class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader): MultivariateGaussian(gaussian[0], gaussian[1]) for gaussian in self.call("gaussians") ] - @property + @property # type: ignore[misc] @since("1.4.0") - def k(self): + def k(self) -> int: """Number of gaussians in mixture.""" return len(self.weights) - def predict(self, x): + @overload + def predict(self, x: "VectorLike") -> np.int64: + ... + + @overload + def predict(self, x: RDD["VectorLike"]) -> RDD[int]: + ... + + def predict(self, x: Union["VectorLike", RDD["VectorLike"]]) -> Union[np.int64, RDD[int]]: """ Find the cluster to which the point 'x' or each point in RDD 'x' has maximum membership in this model. @@ -548,7 +591,17 @@ class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader): z = self.predictSoft(x) return z.argmax() - def predictSoft(self, x): + @overload + def predictSoft(self, x: "VectorLike") -> np.ndarray: + ... + + @overload + def predictSoft(self, x: RDD["VectorLike"]) -> RDD[pyarray.array]: + ... + + def predictSoft( + self, x: Union["VectorLike", RDD["VectorLike"]] + ) -> Union[np.ndarray, RDD[pyarray.array]]: """ Find the membership of point 'x' or each point in RDD 'x' to all mixture components. @@ -579,7 +632,7 @@ class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader): return self.call("predictSoft", _convert_to_vector(x)).toArray() @classmethod - def load(cls, sc, path): + def load(cls, sc: SparkContext, path: str) -> "GaussianMixtureModel": """Load the GaussianMixtureModel from disk. .. versionadded:: 1.5.0 @@ -590,6 +643,8 @@ class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader): path : str Path to where the model is stored. """ + assert sc._jvm is not None + model = cls._load_java(sc, path) wrapper = sc._jvm.org.apache.spark.mllib.api.python.GaussianMixtureModelWrapper(model) return cls(wrapper) @@ -603,7 +658,15 @@ class GaussianMixture: """ @classmethod - def train(cls, rdd, k, convergenceTol=1e-3, maxIterations=100, seed=None, initialModel=None): + def train( + cls, + rdd: RDD["VectorLike"], + k: int, + convergenceTol: float = 1e-3, + maxIterations: int = 100, + seed: Optional[int] = None, + initialModel: Optional[GaussianMixtureModel] = None, + ) -> GaussianMixtureModel: """ Train a Gaussian Mixture clustering model. @@ -658,7 +721,9 @@ class GaussianMixture: return GaussianMixtureModel(java_model) -class PowerIterationClusteringModel(JavaModelWrapper, JavaSaveable, JavaLoader): +class PowerIterationClusteringModel( + JavaModelWrapper, JavaSaveable, JavaLoader["PowerIterationClusteringModel"] +): """ Model produced by :py:class:`PowerIterationClustering`. @@ -711,16 +776,16 @@ class PowerIterationClusteringModel(JavaModelWrapper, JavaSaveable, JavaLoader): ... pass """ - @property + @property # type: ignore[misc] @since("1.5.0") - def k(self): + def k(self) -> int: """ Returns the number of clusters. """ return self.call("k") @since("1.5.0") - def assignments(self): + def assignments(self) -> RDD["PowerIterationClustering.Assignment"]: """ Returns the cluster assignments of this model. """ @@ -728,10 +793,12 @@ class PowerIterationClusteringModel(JavaModelWrapper, JavaSaveable, JavaLoader): @classmethod @since("1.5.0") - def load(cls, sc, path): + def load(cls, sc: SparkContext, path: str) -> "PowerIterationClusteringModel": """ Load a model from the given path. """ + assert sc._jvm is not None + model = cls._load_java(sc, path) wrapper = sc._jvm.org.apache.spark.mllib.api.python.PowerIterationClusteringModelWrapper( model @@ -757,7 +824,13 @@ class PowerIterationClustering: """ @classmethod - def train(cls, rdd, k, maxIterations=100, initMode="random"): + def train( + cls, + rdd: RDD[Tuple[int, int, float]], + k: int, + maxIterations: int = 100, + initMode: str = "random", + ) -> PowerIterationClusteringModel: r""" Train PowerIterationClusteringModel @@ -867,18 +940,20 @@ class StreamingKMeansModel(KMeansModel): 1 """ - def __init__(self, clusterCenters, clusterWeights): + def __init__(self, clusterCenters: List["VectorLike"], clusterWeights: "VectorLike"): super(StreamingKMeansModel, self).__init__(centers=clusterCenters) - self._clusterWeights = list(clusterWeights) + self._clusterWeights = list(clusterWeights) # type: ignore[arg-type] - @property + @property # type: ignore[misc] @since("1.5.0") - def clusterWeights(self): + def clusterWeights(self) -> List[np.float64]: """Return the cluster weights.""" return self._clusterWeights @since("1.5.0") - def update(self, data, decayFactor, timeUnit): + def update( + self, data: RDD["VectorLike"], decayFactor: float, timeUnit: str + ) -> "StreamingKMeansModel": """Update the centroids, according to data .. versionadded:: 1.5.0 @@ -909,7 +984,7 @@ class StreamingKMeansModel(KMeansModel): decayFactor, timeUnit, ) - self.centers = array(updatedModel[0]) + self.centers = array(updatedModel[0]) # type: ignore[assignment] self._clusterWeights = list(updatedModel[1]) return self @@ -938,20 +1013,20 @@ class StreamingKMeans: (default: "batches") """ - def __init__(self, k=2, decayFactor=1.0, timeUnit="batches"): + def __init__(self, k: int = 2, decayFactor: float = 1.0, timeUnit: str = "batches"): self._k = k self._decayFactor = decayFactor if timeUnit not in ["batches", "points"]: raise ValueError("timeUnit should be 'batches' or 'points', got %s." % timeUnit) self._timeUnit = timeUnit - self._model = None + self._model: Optional[StreamingKMeansModel] = None @since("1.5.0") - def latestModel(self): + def latestModel(self) -> Optional[StreamingKMeansModel]: """Return the latest model""" return self._model - def _validate(self, dstream): + def _validate(self, dstream: Any) -> None: if self._model is None: raise ValueError( "Initial centers should be set either by setInitialCenters " "or setRandomCenters." @@ -962,19 +1037,19 @@ class StreamingKMeans: ) @since("1.5.0") - def setK(self, k): + def setK(self, k: int) -> "StreamingKMeans": """Set number of clusters.""" self._k = k return self @since("1.5.0") - def setDecayFactor(self, decayFactor): + def setDecayFactor(self, decayFactor: float) -> "StreamingKMeans": """Set decay factor.""" self._decayFactor = decayFactor return self @since("1.5.0") - def setHalfLife(self, halfLife, timeUnit): + def setHalfLife(self, halfLife: float, timeUnit: str) -> "StreamingKMeans": """ Set number of batches after which the centroids of that particular batch has half the weightage. @@ -984,7 +1059,9 @@ class StreamingKMeans: return self @since("1.5.0") - def setInitialCenters(self, centers, weights): + def setInitialCenters( + self, centers: List["VectorLike"], weights: List[float] + ) -> "StreamingKMeans": """ Set initial centers. Should be set before calling trainOn. """ @@ -992,7 +1069,7 @@ class StreamingKMeans: return self @since("1.5.0") - def setRandomCenters(self, dim, weight, seed): + def setRandomCenters(self, dim: int, weight: float, seed: int) -> "StreamingKMeans": """ Set the initial centers to be random samples from a gaussian population with constant weights. @@ -1000,39 +1077,39 @@ class StreamingKMeans: rng = random.RandomState(seed) clusterCenters = rng.randn(self._k, dim) clusterWeights = tile(weight, self._k) - self._model = StreamingKMeansModel(clusterCenters, clusterWeights) + self._model = StreamingKMeansModel(clusterCenters, clusterWeights) # type: ignore[arg-type] return self @since("1.5.0") - def trainOn(self, dstream): + def trainOn(self, dstream: "DStream[VectorLike]") -> None: """Train the model on the incoming dstream.""" self._validate(dstream) - def update(rdd): - self._model.update(rdd, self._decayFactor, self._timeUnit) + def update(rdd: RDD["VectorLike"]) -> None: + self._model.update(rdd, self._decayFactor, self._timeUnit) # type: ignore[union-attr] dstream.foreachRDD(update) @since("1.5.0") - def predictOn(self, dstream): + def predictOn(self, dstream: "DStream[VectorLike]") -> "DStream[int]": """ Make predictions on a dstream. Returns a transformed dstream object """ self._validate(dstream) - return dstream.map(lambda x: self._model.predict(x)) + return dstream.map(lambda x: self._model.predict(x)) # type: ignore[union-attr] @since("1.5.0") - def predictOnValues(self, dstream): + def predictOnValues(self, dstream: "DStream[Tuple[T, VectorLike]]") -> "DStream[Tuple[T, int]]": """ Make predictions on a keyed dstream. Returns a transformed dstream object. """ self._validate(dstream) - return dstream.mapValues(lambda x: self._model.predict(x)) + return dstream.mapValues(lambda x: self._model.predict(x)) # type: ignore[union-attr] -class LDAModel(JavaModelWrapper, JavaSaveable, Loader): +class LDAModel(JavaModelWrapper, JavaSaveable, Loader["LDAModel"]): """A clustering model derived from the LDA method. @@ -1089,16 +1166,18 @@ class LDAModel(JavaModelWrapper, JavaSaveable, Loader): """ @since("1.5.0") - def topicsMatrix(self): + def topicsMatrix(self) -> np.ndarray: """Inferred topics, where each topic is represented by a distribution over terms.""" return self.call("topicsMatrix").toArray() @since("1.5.0") - def vocabSize(self): + def vocabSize(self) -> int: """Vocabulary size (number of terms or terms in the vocabulary)""" return self.call("vocabSize") - def describeTopics(self, maxTermsPerTopic=None): + def describeTopics( + self, maxTermsPerTopic: Optional[int] = None + ) -> List[Tuple[List[int], List[float]]]: """Return the topics described by weighted terms. .. versionadded:: 1.6.0 @@ -1124,7 +1203,7 @@ class LDAModel(JavaModelWrapper, JavaSaveable, Loader): return topics @classmethod - def load(cls, sc, path): + def load(cls, sc: SparkContext, path: str) -> "LDAModel": """Load the LDAModel from disk. .. versionadded:: 1.5.0 @@ -1153,15 +1232,15 @@ class LDA: @classmethod def train( cls, - rdd, - k=10, - maxIterations=20, - docConcentration=-1.0, - topicConcentration=-1.0, - seed=None, - checkpointInterval=10, - optimizer="em", - ): + rdd: RDD[Tuple[int, "VectorLike"]], + k: int = 10, + maxIterations: int = 20, + docConcentration: float = -1.0, + topicConcentration: float = -1.0, + seed: Optional[int] = None, + checkpointInterval: int = 10, + optimizer: str = "em", + ) -> LDAModel: """Train a LDA model. .. versionadded:: 1.5.0 @@ -1215,7 +1294,7 @@ class LDA: return LDAModel(model) -def _test(): +def _test() -> None: import doctest import numpy import pyspark.mllib.clustering diff --git a/python/pyspark/mllib/clustering.pyi b/python/pyspark/mllib/clustering.pyi deleted file mode 100644 index 8a8401d3565..00000000000 --- a/python/pyspark/mllib/clustering.pyi +++ /dev/null @@ -1,188 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -from typing import overload -from typing import List, NamedTuple, Optional, Tuple, TypeVar - -import array - -from numpy import float64, int64, ndarray -from py4j.java_gateway import JavaObject - -from pyspark.mllib._typing import VectorLike -from pyspark.context import SparkContext -from pyspark.rdd import RDD -from pyspark.mllib.common import JavaModelWrapper -from pyspark.mllib.stat.distribution import MultivariateGaussian -from pyspark.mllib.util import Saveable, Loader, JavaLoader, JavaSaveable -from pyspark.streaming.dstream import DStream - -T = TypeVar("T") - -class BisectingKMeansModel(JavaModelWrapper): - centers: List[VectorLike] - def __init__(self, java_model: JavaObject) -> None: ... - @property - def clusterCenters(self) -> List[ndarray]: ... - @property - def k(self) -> int: ... - @overload - def predict(self, x: VectorLike) -> int: ... - @overload - def predict(self, x: RDD[VectorLike]) -> RDD[int]: ... - @overload - def computeCost(self, x: VectorLike) -> float: ... - @overload - def computeCost(self, x: RDD[VectorLike]) -> float: ... - -class BisectingKMeans: - @classmethod - def train( - self, - rdd: RDD[VectorLike], - k: int = ..., - maxIterations: int = ..., - minDivisibleClusterSize: float = ..., - seed: int = ..., - ) -> BisectingKMeansModel: ... - -class KMeansModel(Saveable, Loader[KMeansModel]): - centers: List[VectorLike] - def __init__(self, centers: List[VectorLike]) -> None: ... - @property - def clusterCenters(self) -> List[ndarray]: ... - @property - def k(self) -> int: ... - @overload - def predict(self, x: VectorLike) -> int: ... - @overload - def predict(self, x: RDD[VectorLike]) -> RDD[int]: ... - def computeCost(self, rdd: RDD[VectorLike]) -> float: ... - def save(self, sc: SparkContext, path: str) -> None: ... - @classmethod - def load(cls, sc: SparkContext, path: str) -> KMeansModel: ... - -class KMeans: - @classmethod - def train( - cls, - rdd: RDD[VectorLike], - k: int, - maxIterations: int = ..., - initializationMode: str = ..., - seed: Optional[int] = ..., - initializationSteps: int = ..., - epsilon: float = ..., - initialModel: Optional[KMeansModel] = ..., - ) -> KMeansModel: ... - -class GaussianMixtureModel(JavaModelWrapper, JavaSaveable, JavaLoader[GaussianMixtureModel]): - @property - def weights(self) -> ndarray: ... - @property - def gaussians(self) -> List[MultivariateGaussian]: ... - @property - def k(self) -> int: ... - @overload - def predict(self, x: VectorLike) -> int64: ... - @overload - def predict(self, x: RDD[VectorLike]) -> RDD[int]: ... - @overload - def predictSoft(self, x: VectorLike) -> ndarray: ... - @overload - def predictSoft(self, x: RDD[VectorLike]) -> RDD[array.array]: ... - @classmethod - def load(cls, sc: SparkContext, path: str) -> GaussianMixtureModel: ... - -class GaussianMixture: - @classmethod - def train( - cls, - rdd: RDD[VectorLike], - k: int, - convergenceTol: float = ..., - maxIterations: int = ..., - seed: Optional[int] = ..., - initialModel: Optional[GaussianMixtureModel] = ..., - ) -> GaussianMixtureModel: ... - -class PowerIterationClusteringModel( - JavaModelWrapper, JavaSaveable, JavaLoader[PowerIterationClusteringModel] -): - @property - def k(self) -> int: ... - def assignments(self) -> RDD[PowerIterationClustering.Assignment]: ... - @classmethod - def load(cls, sc: SparkContext, path: str) -> PowerIterationClusteringModel: ... - -class PowerIterationClustering: - @classmethod - def train( - cls, - rdd: RDD[Tuple[int, int, float]], - k: int, - maxIterations: int = ..., - initMode: str = ..., - ) -> PowerIterationClusteringModel: ... - class Assignment(NamedTuple("Assignment", [("id", int), ("cluster", int)])): ... - -class StreamingKMeansModel(KMeansModel): - def __init__(self, clusterCenters: List[VectorLike], clusterWeights: VectorLike) -> None: ... - @property - def clusterWeights(self) -> List[float64]: ... - centers: List[VectorLike] - def update( - self, data: RDD[VectorLike], decayFactor: float, timeUnit: str - ) -> StreamingKMeansModel: ... - -class StreamingKMeans: - def __init__(self, k: int = ..., decayFactor: float = ..., timeUnit: str = ...) -> None: ... - def latestModel(self) -> StreamingKMeansModel: ... - def setK(self, k: int) -> StreamingKMeans: ... - def setDecayFactor(self, decayFactor: float) -> StreamingKMeans: ... - def setHalfLife(self, halfLife: float, timeUnit: str) -> StreamingKMeans: ... - def setInitialCenters( - self, centers: List[VectorLike], weights: List[float] - ) -> StreamingKMeans: ... - def setRandomCenters(self, dim: int, weight: float, seed: int) -> StreamingKMeans: ... - def trainOn(self, dstream: DStream[VectorLike]) -> None: ... - def predictOn(self, dstream: DStream[VectorLike]) -> DStream[int]: ... - def predictOnValues(self, dstream: DStream[Tuple[T, VectorLike]]) -> DStream[Tuple[T, int]]: ... - -class LDAModel(JavaModelWrapper, JavaSaveable, Loader[LDAModel]): - def topicsMatrix(self) -> ndarray: ... - def vocabSize(self) -> int: ... - def describeTopics( - self, maxTermsPerTopic: Optional[int] = ... - ) -> List[Tuple[List[int], List[float]]]: ... - @classmethod - def load(cls, sc: SparkContext, path: str) -> LDAModel: ... - -class LDA: - @classmethod - def train( - cls, - rdd: RDD[Tuple[int, VectorLike]], - k: int = ..., - maxIterations: int = ..., - docConcentration: float = ..., - topicConcentration: float = ..., - seed: Optional[int] = ..., - checkpointInterval: int = ..., - optimizer: str = ..., - ) -> LDAModel: ... --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org