Cross validation spark example

console output

+--------+--------------------+-----+--------------------+--------------------+--------------------+--------------------+----------+
|category|                text|label|               words|            features|       rawPrediction|         probability|prediction|
+--------+--------------------+-----+--------------------+--------------------+--------------------+--------------------+----------+
|     833|"Чехол-обложка дл...|  1.0|["чехол-обложка, ...|(10000,[514,986,1...|[-379.50617089769...|[3.15784456725782...|       1.0|
|     833|"Чехол-обложка дл...|  1.0|["чехол-обложка, ...|(10000,[290,514,9...|[-395.54097963559...|[3.98362185323457...|       1.0|
|       0|"Держатель для мо...|  0.0|["держатель, для,...|(10000,[34,45,47,...|[-333.85171077966...|[0.88443426309164...|       0.0|
|       9|Шина Nordman RS 1...|  8.0|[шина, nordman, r...|(10000,[1124,1223...|[-70.906588615908...|[5.01470370003123...|       8.0|
|     833|"Набор для зарядк...|  1.0|["набор, для, зар...|(10000,[130,292,5...|[-530.30719860286...|[1.83163510325867...|      14.0|
|       9|Шина Amtel Planet...|  8.0|[шина, amtel, pla...|(10000,[1126,1662...|[-80.650591334952...|[6.82193766816946...|       8.0|
|       9|Шина Dunlop SP SP...|  8.0|[шина, dunlop, sp...|(10000,[829,1124,...|[-95.262588717904...|[2.16493027636750...|       8.0|
|     191|"Переходник для н...| 14.0|["переходник, для...|(10000,[45,469,49...|[-546.51337417465...|[0.99982610988070...|       0.0|
|       0|"Мышь беспроводна...|  0.0|["мышь, беспровод...|(10000,[225,372,1...|[-473.28041687729...|[6.84849843659528...|       3.0|
|     833|"Samsung EB535163...|  1.0|["samsung, eb5351...|(10000,[45,808,98...|[-228.28104225404...|[0.55552587751521...|       0.0|
|     833|Чехол раскладной ...|  1.0|[чехол, раскладно...|(10000,[171,182,1...|[-214.98888433978...|[2.02937182254501...|       1.0|
|       9|Шина Amtel Planet...|  8.0|[шина, amtel, pla...|(10000,[1126,1511...|[-80.047756352839...|[1.38819329810127...|       8.0|
|     833|"Аккумулятор для ...|  1.0|["аккумулятор, дл...|(10000,[33,123,18...|[-480.02930916584...|[1.60510931011430...|       2.0|
|       9|Шина Nokian HAKKA...|  8.0|[шина, nokian, ha...|(10000,[55,1125,1...|[-80.681701728580...|[1.59021876633631...|       8.0|
|     833|"USB дата-кабель ...|  1.0|["usb, дата-кабел...|(10000,[43,45,53,...|[-621.94630029335...|[0.99970045656234...|       0.0|
|     833|"USB дата-кабель ...|  1.0|["usb, дата-кабел...|(10000,[34,43,45,...|[-629.24774463079...|[0.91801668706547...|       0.0|
|       9|Шина Bridgestone ...|  8.0|[шина, bridgeston...|(10000,[393,668,1...|[-85.071625355261...|[5.94073335645149...|       8.0|
|     833|"Чехол-книжка для...|  1.0|["чехол-книжка, д...|(10000,[45,86,261...|[-342.22237461950...|[9.05088066831237...|       1.0|
|     833|"Чехол-футляр для...|  1.0|["чехол-футляр, д...|(10000,[227,262,2...|[-330.84039506220...|[2.08685908911187...|       1.0|
|     833|"Чехол-футляр для...|  1.0|["чехол-футляр, д...|(10000,[227,262,2...|[-332.60097160400...|[5.80838036069557...|       1.0|
+--------+--------------------+-----+--------------------+--------------------+--------------------+--------------------+----------+
only showing top 20 rows

F1 metric = 0.844338

cross-validation-classifier.py

from pyspark.mllib.regression import LabeledPoint
from pyspark.ml.classification import NaiveBayes, MultilayerPerceptronClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.tuning import CrossValidator, CrossValidator
from pyspark.ml.feature import HashingTF, Tokenizer, StringIndexer, Word2Vec, IndexToString
from pyspark.ml import Pipeline
from pyspark.sql import Row

textFile = sc.textFile("/Users/mitallast/Sites/spark/sell.csv")
data = textFile.map(lambda line: line.split(',', 1)).map(lambda p: Row(category=p[0], text=p[1]))
schemaSell = sqlContext.createDataFrame(data)
schemaSell.write.save("/Users/mitallast/Sites/spark/sell.parquet", format="parquet")

schemaSell = sqlContext.read.load("/Users/mitallast/Sites/spark/sell.parquet")

train_data, test_data = schemaSell.randomSplit([0.8, 0.2])

categoryIndexer = StringIndexer(inputCol="category", outputCol="label")
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol="words", outputCol="features", numFeatures=10000)
nb = NaiveBayes(smoothing=1.0, modelType="multinomial")

pipeline = Pipeline(stages=[categoryIndexer, tokenizer, hashingTF, nb])

grid = new ParamGridBuilder()
  .addGrid(hashingTF.numFeatures, [100, 1000, 10000])
  .build()

evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="f1")

cv = CrossValidator(estimator=pipeline, estimatorParamsGrid=grid, evaluator=evaluator, numFolds=100)

model = cv.fit(train_data).bestModel

pr = model.transform(test_data)
metric = evaluator.evaluate(pr)
print "F1 metric = %g" % metric