spark下的贝叶斯模型

贝叶斯模型调用函数比较简单，重点是数据格式的读取。

还是在代码中说明吧

import os
import sys
from numpy import array

from pyspark import SparkConf
from pyspark import SparkContext

#from __future__ import print_function
from operator import add

from pyspark.mllib.feature
import IDF
from pyspark.mllib.feature
import HashingTF
from pyspark.mllib.linalg
import Vectors

from pyspark.mllib.classification
import NaiveBayes, NaiveBayesModel
from pyspark.mllib.regression
import LabeledPoint
//以上是导入各种库

conf = SparkConf()
conf = conf.setMaster("local")
sc = SparkContext(conf=conf)
//以下这两个程序主要是为了把类别转成数字，对别人可能无用，可以借用的部分是，这样返回一个字典。

def split(line):
        return line.split('\t')[1],line.split('\t')[0]

def relation():
        fin = sc.textFile("file:///home/tag2.txt")
        dict_tag = dict(fin.map(split).collect())//返回一个字典。
        return dict_tag
//以下程序的读训练文本和训练的类，分别放在两个文件中，训练模型，并输入测试文本（只有一行）进行分类，返回类标号
def read_train_tag():
        dict_tag = relation()
        fin = sc.textFile("file:///new_train_tag")
        train_tag = fin.map(lambda line :dict_tag[line.strip()]).collect()//将类对应到数字标号
        return train_tag
def tfidf():
        train_tag = read_train_tag()
        doc = sc.textFile("file:///home/new_train_jieba")//分好词的训练文档
        doc1 = doc.map(lambda line:line.split("
"))
        htf = HashingTF(2000)//这里是把特征定义成2000维度，假如文本比较大，可以设置成默认的维度
        tf = htf.transform(doc1)
        idf = IDF()
        model = idf.fit(tf)
        tfidf = model.transform(tf)//构建tfidf模型

        output = tfidf.collect()
        data = []
//由于tfidf返回的是个稀疏矩阵，当时费了好大劲才知道该怎么做。首先是把稀疏矩阵转换成密集矩阵，使用toArray()函数可以把一个稀疏矩阵转成数组，转成数组后倒是可以不用转成密集矩阵了。转成密集矩阵并且加上类别label就是贝叶斯模型想要的格式
        for i
in range(len(output)):
                data.append(LabeledPoint(train_tag[i],Vectors.dense(output[i].toArray())))
        na_model = NaiveBayes.train(sc.parallelize(data))//要改成贝叶斯模型需要的数据格式
        test = sc.textFile("file:///test_data")//输入测试数据就可以进行测试了，当然也要把测试数据改成和训练数据一样的格式

test1 = test.map(lambda line:line.split("
"))
        test_tf = htf.transform(test1)
        test_tfidf = model.transform(test_tf).collect()
        s = na_model.predict(test_tfidf[0].toArray())
        print (s)//是一个类别，后期调研一下，怎么得到每个类别以及概率

tfidf()