Question

Adapted from example in Spark doucmentation.

from pyspark.mllib.classification import LogisticRegressionWithSGD
from pyspark.mllib.regression import LabeledPoint

def parsePoint(md):
    return LabeledPoint(md[-1], md[2])

full_count = mds.count()

# Only use columns with less than 20% missing as features
idxs = [i for i in range(9) if
        mds.filter(lambda p: np.isfinite(p[2][i])).count() > 0.8*full_count]

data = mds.filter(lambda p: np.all(np.isfinite(np.array(p[2])[idxs]))).map(lambda p: parsePoint(p))
train_data, predict_data = data.randomSplit([0.9, 0.1])

model = LogisticRegressionWithSGD.train(train_data)

labelsAndPreds = predict_data.map(lambda p: (p.label, model.predict(p.features)))
trainErr = labelsAndPreds.filter(lambda (v, p): v != p).count() / float(predict_data.count())

print "Training Error = " + str(trainErr)

[0, 2, 4, 5, 6, 7, 8]
5160175 574313
5734488 5160175 574313
Training Error = 0.00356774093569