导入pyspark.conf,pyspark.sparkcontext,pyspark.mllib,实现SVM对于新闻的分类。数据集为多个按照类别分类的文件夹，每个文件夹下为新闻的中文正文内容，采用tf-idf对数据集进行清洗和处理，得到RDD。路径为/project/类别、文本

首先，需要在终端使用以下命令启动pyspark：

pyspark --master yarn --deploy-mode client

然后，在pyspark中进行以下操作：

1. 导入必要的库和模块

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession
from pyspark.ml.feature import HashingTF, IDF, Tokenizer
from pyspark.ml.classification import LinearSVC
from pyspark.ml import Pipeline

2. 创建SparkSession

conf = SparkConf().setAppName("News Classification with SVM")
spark = SparkSession.builder.config(conf=conf).getOrCreate()

3. 加载数据集

path = "/project/*/*"
data = spark.sparkContext.wholeTextFiles(path)

4. 将数据集转换为DataFrame格式

df = data.toDF(["path", "text"])

5. 对文本进行分词和TF-IDF处理

tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="rawFeatures")
idf = IDF(inputCol=hashingTF.getOutputCol(), outputCol="features")

pipeline = Pipeline(stages=[tokenizer, hashingTF, idf])
model = pipeline.fit(df)
result = model.transform(df)

6. 将数据集划分为训练集和测试集

train, test = result.randomSplit([0.8, 0.2], seed=12345)

7. 训练SVM模型并进行预测

svm = LinearSVC(maxIter=10, regParam=0.1)
svmModel = svm.fit(train)
predictions = svmModel.transform(test)

8. 对预测结果进行评估

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

evaluator = MulticlassClassificationEvaluator(predictionCol="prediction")
accuracy = evaluator.evaluate(predictions)
print("Accuracy = %g" % (accuracy))

完整代码如下：

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession
from pyspark.ml.feature import HashingTF, IDF, Tokenizer
from pyspark.ml.classification import LinearSVC
from pyspark.ml import Pipeline
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

conf = SparkConf().setAppName("News Classification with SVM")
spark = SparkSession.builder.config(conf=conf).getOrCreate()

path = "/project/*/*"
data = spark.sparkContext.wholeTextFiles(path)
df = data.toDF(["path", "text"])

tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="rawFeatures")
idf = IDF(inputCol=hashingTF.getOutputCol(), outputCol="features")

pipeline = Pipeline(stages=[tokenizer, hashingTF, idf])
model = pipeline.fit(df)
result = model.transform(df)

train, test = result.randomSplit([0.8, 0.2], seed=12345)

svm = LinearSVC(maxIter=10, regParam=0.1)
svmModel = svm.fit(train)
predictions = svmModel.transform(test)

evaluator = MulticlassClassificationEvaluator(predictionCol="prediction")
accuracy = evaluator.evaluate(predictions)
print("Accuracy = %g" % (accuracy))

spark.stop()