Next, we load a sample graph from a text file as a distributed dataset and package it into `PRVertex` objects. We also cache the distributed dataset because Bagel will use it multiple times and we'd like to avoid recomputing it.

{% highlight scala %}

val numVerts = input.count()

and `DoubleFunction` classes, allowing them to return RDDs of the appropriate

types. Common methods like `filter` and `sample` are implemented by

each specialized RDD class, so filtering a `PairRDD` returns a new `PairRDD`,

framework](http://docs.scala-lang.org/overviews/core/architecture-of-scala-collections.html)).

`Function` classes will need to use `implements` rather than `extends`.

* Certain transformation functions now have multiple versions depending

on the return type. In Spark core, the map functions (`map`, `flatMap`, and

[`mapToPair`](api/java/org/apache/spark/api/java/JavaRDDLike.html#mapToPair(org.apache.spark.api.java.function.PairFunction))

and [`mapToDouble`](api/java/org/apache/spark/api/java/JavaRDDLike.html#mapToDouble(org.apache.spark.api.java.function.DoubleFunction)).

Spark Streaming also uses the same approach, e.g. [`transformToPair`](api/java/org/apache/spark/streaming/api/java/JavaDStreamLike.html#transformToPair(org.apache.spark.api.java.function.Function)).

import org.apache.spark.api.java.*;

import org.apache.spark.api.java.function.*;

JavaRDD<String> words = lines.flatMap(

new FlatMapFunction<String, String>() {

return Arrays.asList(s.split(" "));

}

}

{% highlight java %}

class Split extends FlatMapFunction<String, String> {

return Arrays.asList(s.split(" "));

}

JavaRDD<String> words = lines.flatMap(new Split());

{% endhighlight %}

import scala.Tuple2;

JavaPairRDD<String, Integer> ones = words.mapToPair(

new PairFunction<String, String, Integer>() {

}

}

);

{% highlight java %}

JavaPairRDD<String, Integer> counts = ones.reduceByKey(

new Function2<Integer, Integer, Integer>() {

return i1 + i2;

}

}

MLlib supports local vectors and matrices stored on a single machine,

as well as distributed matrices backed by one or more RDDs.

In the current implementation, local vectors and matrices are simple data models

[Breeze](http://www.scalanlp.org/) and [jblas](http://jblas.org/).

A training example used in supervised learning is called "labeled point" in MLlib.

import org.apache.spark.mllib.util.MLUtils;

import org.apache.spark.rdd.RDDimport;

{% endhighlight %}

</div>

</div>

created from a `JavaRDD<Vector>` instance. Then we can compute its column summary statistics.

{% highlight java %}

import org.apache.spark.mllib.linalg.Vector;

import org.apache.spark.mllib.linalg.distributed.RowMatrix;

val summary: MultivariateStatisticalSummary = mat.computeColumnSummaryStatistics()

println(summary.mean) // a dense vector containing the mean value for each column

println(summary.variance) // column-wise variance

// Compute the covariance matrix.

{% endhighlight %}

</div>

</div>

its row indices.

{% highlight java %}

import org.apache.spark.mllib.linalg.distributed.IndexedRow;

import org.apache.spark.mllib.linalg.distributed.IndexedRowMatrix;

import org.apache.spark.mllib.linalg.distributed.RowMatrix;

// Create an IndexedRowMatrix from a JavaRDD<IndexedRow>.

IndexedRowMatrix mat = new IndexedRowMatrix(rows.rdd());

with sparse rows by calling `toIndexedRowMatrix`.

{% highlight scala %}

import org.apache.spark.mllib.linalg.distributed.CoordinateMatrix;

import org.apache.spark.mllib.linalg.distributed.MatrixEntry;

JavaRDD<MatrixEntry> entries = ... // a JavaRDD of matrix entries

MLlib supports

[k-means](http://en.wikipedia.org/wiki/K-means_clustering) clustering, one of

the most commonly used clustering algorithms that clusters the data points into

variant of the [k-means++](http://en.wikipedia.org/wiki/K-means%2B%2B) method

called [kmeans||](http://theory.stanford.edu/~sergei/papers/vldb12-kmpar.pdf).

The implementation in MLlib has the following parameters:

* *runs* is the number of times to run the k-means algorithm (k-means is not

guaranteed to find a globally optimal solution, and when run multiple times on

a given dataset, the algorithm returns the best clustering result).

* *epsilon* determines the distance threshold within which we consider k-means to have converged.

}.join(predictions)

val MSE = ratesAndPreds.map{

case ((user, product), (r1, r2)) => math.pow((r1- r2), 2)

println("Mean Squared Error = " + MSE)

{% endhighlight %}

[Elements of Statistical Machine Learning](http://statweb.stanford.edu/~tibs/ElemStatLearn/) for

details). For example, for a binary classification problem with one categorical feature with three

categories A, B and C with corresponding proportion of label 1 as 0.2, 0.6 and 0.4, the categorical

and A , B \| C where \| denotes the split.

The recursive tree construction is stopped at a node when one of the two conditions is met:

2. No split candidate leads to an information gain at the node.

val prediction = model.predict(point.features)

(point.label, prediction)

}

println("training Mean Squared Error = " + MSE)

{% endhighlight %}

</div>

<div class="codetabs">

<div data-lang="scala" markdown="1">

{% highlight scala %}

val mat: RowMatrix = ...

// Compute the top 20 singular values and corresponding singular vectors.

The number of columns should be small, e.g, less than 1000.

{% highlight scala %}

val mat: RowMatrix = ...

// Compute the top 10 principal components.

import org.apache.spark.mllib.linalg.Vectors;

double[] array = ... // a double array

{% endhighlight %}

[`Vectors`](api/mllib/index.html#org.apache.spark.mllib.linalg.Vectors$) provides factory methods to

import org.apache.spark.mllib.optimization.L1Updater

val svmAlg = new SVMWithSGD()

{% endhighlight %}

Similarly, you can use replace `SVMWithSGD` by

val prediction = model.predict(point.features)

(point.label, prediction)

}

println("training Mean Squared Error = " + MSE)

{% endhighlight %}

between every pair of features. Naive Bayes can be trained very efficiently. Within a single pass to

the training data, it computes the conditional probability distribution of each feature given label,

and then it applies Bayes' theorem to compute the conditional probability distribution of label

[Naive Bayes classifier](http://en.wikipedia.org/wiki/Naive_Bayes_classifier).

In MLlib, we implemented multinomial naive Bayes, which is typically used for document

classification. Within that context, each observation is a document, each feature represents a term,

or the section

[Naive Bayes text classification](http://nlp.stanford.edu/IR-book/html/htmledition/naive-bayes-text-classification-1.html)

from the book Introduction to Information

can be used for evaluation and prediction.

{% highlight java %}

import org.apache.spark.mllib.classification.NaiveBayes;

JavaRDD<LabeledPoint> training = ... // training set

JavaRDD<LabeledPoint> test = ... // test set

NaiveBayesModel model = NaiveBayes.train(training.rdd(), 1.0);

}

JavaPairRDD<Double, Double> predictionAndLabel =

prediction.zip(test.map(new Function<LabeledPoint, Double>() {

return p.label();

}

double accuracy = 1.0 * predictionAndLabel.filter(new Function<Tuple2<Double, Double>, Boolean>() {

return pl._1() == pl._2();

}

{% endhighlight %}

</div>