在pyspark mllib上，实现SVM对于新闻的分类。数据集为多个按新闻类型分类的文件夹，每个文件夹下为对应类别的新闻正文文本。如何通过sparkcontext获取数据集，转换成RDD形式。要求对正文的数据集进行tf-idf处理，数据清洗，停用词处理等。同时注意中文的乱码问题

首先，我们需要导入必要的库：

from pyspark import SparkContext
from pyspark.mllib.feature import HashingTF, IDF
from pyspark.mllib.regression import LabeledPoint
import jieba
import re

然后，我们可以使用SparkContext来获取数据集并转换成RDD形式：

sc = SparkContext(appName="SVM News Classification")

# 读取多个文件夹下的新闻文本并转换成RDD形式
data = sc.wholeTextFiles("path/to/news/folders/*")

在读取文本文件时，我们可以指定编码方式来解决中文乱码问题：

data = sc.wholeTextFiles("path/to/news/folders/*", use_unicode=False, charset="gbk")

下一步是对正文数据进行清洗和停用词处理。我们可以定义一个函数来完成这些任务：

def clean_text(text):
    # 去除标点符号
    text = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9]", " ", text)
    # 中文分词
    words = jieba.cut(text)
    # 去除停用词
    stopwords = set(['的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'])
    words = [word for word in words if word not in stopwords]
    # 返回清洗后的文本
    return " ".join(words)

接下来，我们可以使用Spark的map函数来对每个文本进行清洗和处理：

# 对每个文本进行清洗和停用词处理
cleaned_data = data.map(lambda x: (x[0], clean_text(x[1])))

现在，我们可以使用HashingTF和IDF方法将文本转换成tf-idf向量：

# 定义HashingTF和IDF对象
hashingTF = HashingTF()
tf = hashingTF.transform(cleaned_data.map(lambda x: x[1].split()))
tf.cache()
idf = IDF().fit(tf)
# 转换成tf-idf向量
tfidf = idf.transform(tf)

最后，我们需要将文本标签和tf-idf向量转换成LabeledPoint形式：

# 获取文本标签
labels = cleaned_data.map(lambda x: x[0].split("/")[-2])
# 将tf-idf向量和标签转换成LabeledPoint形式
data_labeled = labels.zip(tfidf).map(lambda x: LabeledPoint(x[0], x[1]))

现在，我们可以使用SVM算法对新闻进行分类：

from pyspark.mllib.classification import SVMWithSGD, SVMModel

# 将数据集分为训练集和测试集
(training_data, test_data) = data_labeled.randomSplit([0.7, 0.3])

# 训练SVM模型
model = SVMWithSGD.train(training_data, iterations=100)

# 对测试集进行预测
predictions = model.predict(test_data.map(lambda x: x.features))
labels_and_predictions = test_data.map(lambda lp: lp.label).zip(predictions)

# 计算模型的准确率
accuracy = labels_and_predictions.filter(lambda x: x[0] == x[1]).count() / float(test_data.count())
print("Accuracy = " + str(accuracy))

这样，我们就完成了基于pyspark mllib的SVM新闻分类器的实现。