<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Machine Learning in Spark
Data Types
MLLIB support RDD of numpy arrays, vectors, matrices, labeled points.
Supported Distributed Vectors
Step1:  Example Labeled Points
Step2:  Example Creating a Word-count RDD
MLLIB Classification Package
Naive Bayes
NB.train(Xrdd_train)
NB.train(Xrdd_test.features)
model.theta returns $\log P(\text{attribute | class})$
model.pi returns $\log$ of class priors $\log (\text{class})$
Step3:  Confusion matrix
Step4:  Decision Tree
<ASSISTANT_TASK:>
Python Code:
from pyspark import SparkContext 
sc = SparkContext('local','example')
from pyspark.mllib.linalg import Vectors
x = Vectors.dense([1,2,3,4])
x[0]
x = [Vectors.dense([1,2,3,4,5]), Vectors.dense([6,7,8,9,10])]
xrdd = sc.parallelize(x, 2)
xrdd.glom().collect()
Explanation: Machine Learning in Spark
Data Types
MLLIB support RDD of numpy arrays, vectors, matrices, labeled points.
Supported Distributed Vectors:
  * dense vectors
  * sparse vectors
End of explanation
from pyspark.mllib.regression import LabeledPoint as LP
pt = LP(1, Vectors.dense(2,-1,4))
print("Label: ", pt.label)
print("Feature Vector: ", pt.features)
Explanation: Example Labeled Points
End of explanation
from pysparl.mllib.classification import NaiveBayes as NB
nbmodel = NB.train(Xrdd_train)
testpred = NB.train(Xrdd_test.features)
Explanation: Example Creating a Word-count RDD
MLLIB Classification Package
Naive Bayes
NB.train(Xrdd_train)
NB.train(Xrdd_test.features)
model.theta returns $\log P(\text{attribute | class})$
model.pi returns $\log$ of class priors $\log (\text{class})$
End of explanation
trainpred = nbmodel.predict(Xrdd_train.features)
cf_mat [Xrdd_train.label][trainpred] += 1
Explanation: Confusion matrix
End of explanation
from pyspark.mllib.classification import DecisionTree
dtmodel = DecisionTree.trainClassifier(Xrdd_train, 
                                       numClasses = 2,
                                       impurity = 'entropy', ## options: gini or entropy
                                       maxDepth = 5,
                                       maxBins = 32,
                                       minInstancesPerNode = 2)
Explanation: Decision Tree
End of explanation
<END_TASK>