From 2940201ad86e5dee16cf7386b3c934fc75c15582 Mon Sep 17 00:00:00 2001 From: Tor Myklebust Date: Fri, 20 Dec 2013 01:33:32 -0500 Subject: Tests for the Python side of the mllib bindings. --- python/pyspark/mllib.py | 224 +++++++++++++++++++++++++++++++++++++----------- 1 file changed, 172 insertions(+), 52 deletions(-) (limited to 'python') diff --git a/python/pyspark/mllib.py b/python/pyspark/mllib.py index 21f3c0312c..aa9fc76c29 100644 --- a/python/pyspark/mllib.py +++ b/python/pyspark/mllib.py @@ -1,4 +1,5 @@ from numpy import * +from pyspark import SparkContext # Double vector format: # @@ -7,44 +8,106 @@ from numpy import * # Double matrix format: # # [8-byte 2] [8-byte rows] [8-byte cols] [rows*cols*8 bytes of data] -# +# # This is all in machine-endian. That means that the Java interpreter and the # Python interpreter must agree on what endian the machine is. def _deserialize_byte_array(shape, ba, offset): + """Wrapper around ndarray aliasing hack. + + >>> x = array([1.0, 2.0, 3.0, 4.0, 5.0]) + >>> array_equal(x, _deserialize_byte_array(x.shape, x.data, 0)) + True + >>> x = array([1.0, 2.0, 3.0, 4.0]).reshape(2,2) + >>> array_equal(x, _deserialize_byte_array(x.shape, x.data, 0)) + True + """ ar = ndarray(shape=shape, buffer=ba, offset=offset, dtype="float64", order='C') return ar.copy() def _serialize_double_vector(v): - if (type(v) == ndarray and v.dtype == float64 and v.ndim == 1): - length = v.shape[0] - ba = bytearray(16 + 8*length) - header = ndarray(shape=[2], buffer=ba, dtype="int64") - header[0] = 1 - header[1] = length - copyto(ndarray(shape=[length], buffer=ba, offset=16, - dtype="float64"), v) - return ba - else: - raise TypeError("_serialize_double_vector called on a " - "non-double-vector") + """Serialize a double vector into a mutually understood format. + + >>> _serialize_double_vector(array([])) + bytearray(b'\\x01\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00') + >>> _serialize_double_vector(array([0.0, 1.0])) + bytearray(b'\\x01\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x02\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\xf0?') + >>> _serialize_double_vector("hello, world") + Traceback (most recent call last): + File "/usr/lib/python2.7/doctest.py", line 1289, in __run + compileflags, 1) in test.globs + File "", line 1, in + _serialize_double_vector("hello, world") + File "python/pyspark/mllib.py", line 41, in _serialize_double_vector + raise TypeError("_serialize_double_vector called on a %s; wanted ndarray" % type(v)) + TypeError: _serialize_double_vector called on a ; wanted ndarray + >>> _serialize_double_vector(array([0, 1])) + Traceback (most recent call last): + File "/usr/lib/python2.7/doctest.py", line 1289, in __run + compileflags, 1) in test.globs + File "", line 1, in + _serialize_double_vector(array([0, 1])) + File "python/pyspark/mllib.py", line 51, in _serialize_double_vector + "wanted ndarray of float64" % v.dtype) + TypeError: _serialize_double_vector called on an ndarray of int64; wanted ndarray of float64 + >>> _serialize_double_vector(array([0.0, 1.0, 2.0, 3.0]).reshape(2,2)) + Traceback (most recent call last): + File "/usr/lib/python2.7/doctest.py", line 1289, in __run + compileflags, 1) in test.globs + File "", line 1, in + _serialize_double_vector(array([0.0, 1.0, 2.0, 3.0]).reshape(2,2)) + File "python/pyspark/mllib.py", line 62, in _serialize_double_vector + "wanted a 1darray" % v.ndim) + TypeError: _serialize_double_vector called on a 2darray; wanted a 1darray + """ + if type(v) != ndarray: + raise TypeError("_serialize_double_vector called on a %s; " + "wanted ndarray" % type(v)) + if v.dtype != float64: + raise TypeError("_serialize_double_vector called on an ndarray of %s; " + "wanted ndarray of float64" % v.dtype) + if v.ndim != 1: + raise TypeError("_serialize_double_vector called on a %ddarray; " + "wanted a 1darray" % v.ndim) + length = v.shape[0] + ba = bytearray(16 + 8*length) + header = ndarray(shape=[2], buffer=ba, dtype="int64") + header[0] = 1 + header[1] = length + copyto(ndarray(shape=[length], buffer=ba, offset=16, + dtype="float64"), v) + return ba def _deserialize_double_vector(ba): - if (type(ba) == bytearray and len(ba) >= 16 and (len(ba) & 7 == 0)): - header = ndarray(shape=[2], buffer=ba, dtype="int64") - if (header[0] != 1): - raise TypeError("_deserialize_double_vector called on bytearray " - "with wrong magic") - length = header[1] - if (len(ba) != 8*length + 16): - raise TypeError("_deserialize_double_vector called on bytearray " - "with wrong length") - return _deserialize_byte_array([length], ba, 16) - else: - raise TypeError("_deserialize_double_vector called on a non-bytearray") + """Deserialize a double vector from a mutually understood format. + + >>> x = array([1.0, 2.0, 3.0, 4.0, -1.0, 0.0, -0.0]) + >>> array_equal(x, _deserialize_double_vector(_serialize_double_vector(x))) + True + """ + if type(ba) != bytearray: + raise TypeError("_deserialize_double_vector called on a %s; " + "wanted bytearray" % type(ba)) + if len(ba) < 16: + raise TypeError("_deserialize_double_vector called on a %d-byte array, " + "which is too short" % len(ba)) + if (len(ba) & 7) != 0: + raise TypeError("_deserialize_double_vector called on a %d-byte array, " + "which is not a multiple of 8" % len(ba)) + header = ndarray(shape=[2], buffer=ba, dtype="int64") + if header[0] != 1: + raise TypeError("_deserialize_double_vector called on bytearray " + "with wrong magic") + length = header[1] + if len(ba) != 8*length + 16: + raise TypeError("_deserialize_double_vector called on bytearray " + "with wrong length") + return _deserialize_byte_array([length], ba, 16) def _serialize_double_matrix(m): + """Serialize a double matrix into a mutually understood format. + """ if (type(m) == ndarray and m.dtype == float64 and m.ndim == 2): rows = m.shape[0] cols = m.shape[1] @@ -61,22 +124,31 @@ def _serialize_double_matrix(m): "non-double-matrix") def _deserialize_double_matrix(ba): - if (type(ba) == bytearray and len(ba) >= 24 and (len(ba) & 7 == 0)): - header = ndarray(shape=[3], buffer=ba, dtype="int64") - if (header[0] != 2): - raise TypeError("_deserialize_double_matrix called on bytearray " - "with wrong magic") - rows = header[1] - cols = header[2] - if (len(ba) != 8*rows*cols + 24): - raise TypeError("_deserialize_double_matrix called on bytearray " - "with wrong length") - return _deserialize_byte_array([rows, cols], ba, 24) - else: - raise TypeError("_deserialize_double_matrix called on a non-bytearray") + """Deserialize a double matrix from a mutually understood format. + """ + if type(ba) != bytearray: + raise TypeError("_deserialize_double_matrix called on a %s; " + "wanted bytearray" % type(ba)) + if len(ba) < 24: + raise TypeError("_deserialize_double_matrix called on a %d-byte array, " + "which is too short" % len(ba)) + if (len(ba) & 7) != 0: + raise TypeError("_deserialize_double_matrix called on a %d-byte array, " + "which is not a multiple of 8" % len(ba)) + header = ndarray(shape=[3], buffer=ba, dtype="int64") + if (header[0] != 2): + raise TypeError("_deserialize_double_matrix called on bytearray " + "with wrong magic") + rows = header[1] + cols = header[2] + if (len(ba) != 8*rows*cols + 24): + raise TypeError("_deserialize_double_matrix called on bytearray " + "with wrong length") + return _deserialize_byte_array([rows, cols], ba, 24) def _linear_predictor_typecheck(x, coeffs): - """Predict the class of the vector x.""" + """Check that x is a one-dimensional vector of the right shape. + This is a temporary hackaround until I actually implement bulk predict.""" if type(x) == ndarray: if x.ndim == 1: if x.shape == coeffs.shape: @@ -98,12 +170,17 @@ class LinearModel(object): self._intercept = intercept class LinearRegressionModelBase(LinearModel): - """A linear regression model.""" + """A linear regression model. + + >>> lrmb = LinearRegressionModelBase(array([1.0, 2.0]), 0.1) + >>> abs(lrmb.predict(array([-1.03, 7.777])) - 14.624) < 1e-6 + True + """ def predict(self, x): """Predict the value of the dependent variable given a vector x""" """containing values for the independent variables.""" - _linear_predictor_typecheck(x, _coeff) - return dot(_coeff, x) + _intercept + _linear_predictor_typecheck(x, self._coeff) + return dot(self._coeff, x) + self._intercept # Map a pickled Python RDD of numpy double vectors to a Java RDD of # _serialized_double_vectors @@ -145,7 +222,11 @@ def _regression_train_wrapper(sc, train_func, klass, data, initial_weights): return klass(_deserialize_double_vector(ans[0]), ans[1]); class LinearRegressionModel(LinearRegressionModelBase): - """A linear regression model derived from a least-squares fit.""" + """A linear regression model derived from a least-squares fit. + + >>> data = array([0.0, 0.0, 1.0, 1.0, 3.0, 2.0, 2.0, 3.0]).reshape(4,2) + >>> lrm = LinearRegressionModel.train(sc, sc.parallelize(data), initial_weights=array([1.0])) + """ @classmethod def train(cls, sc, data, iterations=100, step=1.0, mini_batch_fraction=1.0, initial_weights=None): @@ -156,8 +237,12 @@ class LinearRegressionModel(LinearRegressionModelBase): LinearRegressionModel, data, initial_weights) class LassoModel(LinearRegressionModelBase): - """A linear regression model derived from a least-squares fit with an """ - """l_1 penalty term.""" + """A linear regression model derived from a least-squares fit with an + l_1 penalty term. + + >>> data = array([0.0, 0.0, 1.0, 1.0, 3.0, 2.0, 2.0, 3.0]).reshape(4,2) + >>> lrm = LassoModel.train(sc, sc.parallelize(data), initial_weights=array([1.0])) + """ @classmethod def train(cls, sc, data, iterations=100, step=1.0, reg_param=1.0, mini_batch_fraction=1.0, initial_weights=None): @@ -168,8 +253,12 @@ class LassoModel(LinearRegressionModelBase): LassoModel, data, initial_weights) class RidgeRegressionModel(LinearRegressionModelBase): - """A linear regression model derived from a least-squares fit with an """ - """l_2 penalty term.""" + """A linear regression model derived from a least-squares fit with an + l_2 penalty term. + + >>> data = array([0.0, 0.0, 1.0, 1.0, 3.0, 2.0, 2.0, 3.0]).reshape(4,2) + >>> lrm = RidgeRegressionModel.train(sc, sc.parallelize(data), initial_weights=array([1.0])) + """ @classmethod def train(cls, sc, data, iterations=100, step=1.0, reg_param=1.0, mini_batch_fraction=1.0, initial_weights=None): @@ -180,7 +269,11 @@ class RidgeRegressionModel(LinearRegressionModelBase): RidgeRegressionModel, data, initial_weights) class LogisticRegressionModel(LinearModel): - """A linear binary classification model derived from logistic regression.""" + """A linear binary classification model derived from logistic regression. + + >>> data = array([0.0, 0.0, 1.0, 1.0, 1.0, 2.0, 1.0, 3.0]).reshape(4,2) + >>> lrm = LogisticRegressionModel.train(sc, sc.parallelize(data)) + """ def predict(self, x): _linear_predictor_typecheck(x, _coeff) margin = dot(x, _coeff) + intercept @@ -197,7 +290,11 @@ class LogisticRegressionModel(LinearModel): LogisticRegressionModel, data, initial_weights) class SVMModel(LinearModel): - """A support vector machine.""" + """A support vector machine. + + >>> data = array([0.0, 0.0, 1.0, 1.0, 1.0, 2.0, 1.0, 3.0]).reshape(4,2) + >>> svm = SVMModel.train(sc, sc.parallelize(data)) + """ def predict(self, x): _linear_predictor_typecheck(x, _coeff) margin = dot(x, _coeff) + intercept @@ -212,15 +309,24 @@ class SVMModel(LinearModel): SVMModel, data, initial_weights) class KMeansModel(object): - """A clustering model derived from the k-means method.""" + """A clustering model derived from the k-means method. + + >>> data = array([0.0, 0.0, 1.0,1.0, 9.0,8.0, 8.0,9.0]).reshape(4,2) + >>> clusters = KMeansModel.train(sc, sc.parallelize(data), 2, maxIterations=10, runs=30, initialization_mode="random") + >>> clusters.predict(array([0.0, 0.0])) == clusters.predict(array([1.0, 1.0])) + True + >>> clusters.predict(array([8.0, 9.0])) == clusters.predict(array([9.0, 8.0])) + True + >>> clusters = KMeansModel.train(sc, sc.parallelize(data), 2) + """ def __init__(self, centers_): self.centers = centers_ def predict(self, x): best = 0 best_distance = 1e75 - for i in range(0, centers.shape[0]): - diff = x - centers[i] + for i in range(0, self.centers.shape[0]): + diff = x - self.centers[i] distance = sqrt(dot(diff, diff)) if distance < best_distance: best = i @@ -239,3 +345,17 @@ class KMeansModel(object): raise RuntimeError("JVM call result had first element of type " + type(ans[0]) + " which is not bytearray"); return KMeansModel(_deserialize_double_matrix(ans[0])); + +def _test(): + import doctest + globs = globals().copy() + globs['sc'] = SparkContext('local[4]', 'PythonTest', batchSize=2) + (failure_count, test_count) = doctest.testmod(globs=globs, + optionflags=doctest.ELLIPSIS) + globs['sc'].stop() + print failure_count,"failures among",test_count,"tests" + if failure_count: + exit(-1) + +if __name__ == "__main__": + _test() -- cgit v1.2.3