Source Code for Module pyspark.mllib.classification

  1  # 
  2  # Licensed to the Apache Software Foundation (ASF) under one or more 
  3  # contributor license agreements.  See the NOTICE file distributed with 
  4  # this work for additional information regarding copyright ownership. 
  5  # The ASF licenses this file to You under the Apache License, Version 2.0 
  6  # (the "License"); you may not use this file except in compliance with 
  7  # the License.  You may obtain a copy of the License at 
  8  # 
  9  #    http://www.apache.org/licenses/LICENSE-2.0 
 10  # 
 11  # Unless required by applicable law or agreed to in writing, software 
 12  # distributed under the License is distributed on an "AS IS" BASIS, 
 13  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 14  # See the License for the specific language governing permissions and 
 15  # limitations under the License. 
 16  # 
 17   
 18  import numpy 
 19   
 20  from numpy import array, dot, shape 
 21  from pyspark import SparkContext 
 22  from pyspark.mllib._common import \ 
 23      _get_unmangled_rdd, _get_unmangled_double_vector_rdd, \ 
 24      _serialize_double_matrix, _deserialize_double_matrix, \ 
 25      _serialize_double_vector, _deserialize_double_vector, \ 
 26      _get_initial_weights, _serialize_rating, _regression_train_wrapper, \ 
 27      LinearModel, _linear_predictor_typecheck 
 28  from math import exp, log 
 29   
 30 -class LogisticRegressionModel(LinearModel): 
 31      """A linear binary classification model derived from logistic regression. 
 32   
 33      >>> data = array([0.0, 0.0, 1.0, 1.0, 1.0, 2.0, 1.0, 3.0]).reshape(4,2) 
 34      >>> lrm = LogisticRegressionWithSGD.train(sc.parallelize(data)) 
 35      >>> lrm.predict(array([1.0])) > 0 
 36      True 
 37      """ 
 38 -    def predict(self, x): 
 39          _linear_predictor_typecheck(x, self._coeff) 
 40          margin = dot(x, self._coeff) + self._intercept 
 41          prob = 1/(1 + exp(-margin)) 
 42          return 1 if prob > 0.5 else 0 
 43   
 44 -class LogisticRegressionWithSGD(object): 
 45      @classmethod 
 46 -    def train(cls, data, iterations=100, step=1.0, 
 47                miniBatchFraction=1.0, initialWeights=None): 
 48          """Train a logistic regression model on the given data.""" 
 49          sc = data.context 
 50          return _regression_train_wrapper(sc, lambda d, i: 
 51                  sc._jvm.PythonMLLibAPI().trainLogisticRegressionModelWithSGD(d._jrdd, 
 52                          iterations, step, miniBatchFraction, i), 
 53                  LogisticRegressionModel, data, initialWeights) 
 54   
 55 -class SVMModel(LinearModel): 
 56      """A support vector machine. 
 57   
 58      >>> data = array([0.0, 0.0, 1.0, 1.0, 1.0, 2.0, 1.0, 3.0]).reshape(4,2) 
 59      >>> svm = SVMWithSGD.train(sc.parallelize(data)) 
 60      >>> svm.predict(array([1.0])) > 0 
 61      True 
 62      """ 
 63 -    def predict(self, x): 
 64          _linear_predictor_typecheck(x, self._coeff) 
 65          margin = dot(x, self._coeff) + self._intercept 
 66          return 1 if margin >= 0 else 0 
 67   
 68 -class SVMWithSGD(object): 
 69      @classmethod 
 70 -    def train(cls, data, iterations=100, step=1.0, regParam=1.0, 
 71                miniBatchFraction=1.0, initialWeights=None): 
 72          """Train a support vector machine on the given data.""" 
 73          sc = data.context 
 74          return _regression_train_wrapper(sc, lambda d, i: 
 75                  sc._jvm.PythonMLLibAPI().trainSVMModelWithSGD(d._jrdd, 
 76                          iterations, step, regParam, miniBatchFraction, i), 
 77                  SVMModel, data, initialWeights) 
 78   
 79 -class NaiveBayesModel(object): 
 80      """ 
 81      Model for Naive Bayes classifiers. 
 82   
 83      Contains two parameters: 
 84      - pi: vector of logs of class priors (dimension C) 
 85      - theta: matrix of logs of class conditional probabilities (CxD) 
 86   
 87      >>> data = array([0.0, 0.0, 1.0, 0.0, 0.0, 2.0, 1.0, 1.0, 0.0]).reshape(3,3) 
 88      >>> model = NaiveBayes.train(sc.parallelize(data)) 
 89      >>> model.predict(array([0.0, 1.0])) 
 90      0 
 91      >>> model.predict(array([1.0, 0.0])) 
 92      1 
 93      """ 
 94   
 95 -    def __init__(self, pi, theta): 
 96          self.pi = pi 
 97          self.theta = theta 
 98   
 99 -    def predict(self, x): 
100          """Return the most likely class for a data vector x""" 
101          return numpy.argmax(self.pi + dot(x, self.theta)) 
102   
103 -class NaiveBayes(object): 
104      @classmethod 
105 -    def train(cls, data, lambda_=1.0): 
106          """ 
107          Train a Naive Bayes model given an RDD of (label, features) vectors. 
108   
109          This is the Multinomial NB (U{http://tinyurl.com/lsdw6p}) which can 
110          handle all kinds of discrete data.  For example, by converting 
111          documents into TF-IDF vectors, it can be used for document 
112          classification.  By making every vector a 0-1 vector, it can also be 
113          used as Bernoulli NB (U{http://tinyurl.com/p7c96j6}). 
114   
115          @param data: RDD of NumPy vectors, one per element, where the first 
116                 coordinate is the label and the rest is the feature vector 
117                 (e.g. a count vector). 
118          @param lambda_: The smoothing parameter 
119          """ 
120          sc = data.context 
121          dataBytes = _get_unmangled_double_vector_rdd(data) 
122          ans = sc._jvm.PythonMLLibAPI().trainNaiveBayes(dataBytes._jrdd, lambda_) 
123          return NaiveBayesModel( 
124              _deserialize_double_vector(ans[0]), 
125              _deserialize_double_matrix(ans[1])) 
126   
127   
128 -def _test(): 
129      import doctest 
130      globs = globals().copy() 
131      globs['sc'] = SparkContext('local[4]', 'PythonTest', batchSize=2) 
132      (failure_count, test_count) = doctest.testmod(globs=globs, 
133              optionflags=doctest.ELLIPSIS) 
134      globs['sc'].stop() 
135      if failure_count: 
136          exit(-1) 
137   
138  if __name__ == "__main__": 
139      _test() 
140