01

引言

在本文，我们将给大家带来如何使用PySpark处理文本数据。文本数据可以是结构化或者非结构化的，我们必须进行多部处理，才能把数据处理成可供模型学习的输入数据。

2

分词

NLP的底层任务由易到难大致可以分为词法分析、句法分析和语义分析。分词是词法分析（还包括词性标注和命名实体识别）中最基本的任务。首先需要创建spark对象，并创建一个spark的DataFrame。

from pyspark.sql import SparkSession
spark=SparkSession.builder.appName('nlp').getOrCreate()
df=spark.createDataFrame([(1,'I really liked this movie'),
                         (2,'I would recommend this movie to my friends'),
                         (3,'movie was alright but acting was horrible'),
                         (4,'I am never watching that movie ever again')],
                        ['user_id','review'])
df.show(5,False)
+-------+------------------------------------------+
|user_id|review                                    |
+-------+------------------------------------------+
|1      |I really liked this movie                 |
|2      |I would recommend this movie to my friends|
|3      |movie was alright but acting was horrible |
|4      |I am never watching that movie ever again |
+-------+------------------------------------------+

下一步需要从Spark库中引入Tokenizer。然后必须传入输入列并对标记之后的输出列进行命名。这里使用transform函数，以便将标记应用到review列。可以看到tokens列就是已经分好词的文本。

from pyspark.ml.feature import Tokenizer
tokenization=Tokenizer(inputCol='review',outputCol='tokens')
tokenized_df=tokenization.transform(df)
tokenized_df.show(4,False)
+-------+------------------------------------------+---------------------------------------------------+
|user_id|review                                    |tokens                                             |
+-------+------------------------------------------+---------------------------------------------------+
|1      |I really liked this movie                 |[i, really, liked, this, movie]                    |
|2      |I would recommend this movie to my friends|[i, would, recommend, this, movie, to, my, friends]|
|3      |movie was alright but acting was horrible |[movie, was, alright, but, acting, was, horrible]  |
|4      |I am never watching that movie ever again |[i, am, never, watching, that, movie, ever, again] |
+-------+------------------------------------------+---------------------------------------------------+

3

去除停用词

我们可以看到tokens中包含很多通用的单词，例如the、to等。这些单词被称为停用词，这些词对于分析没有多大的用处，并且会增大计算开销，因此我们在进行下游任务时，会考虑将这些词从文本中去除。这便用到Spark中的StopWordsRemover。

from pyspark.ml.feature import StopWordsRemover
stopword_removal=StopWordsRemover(inputCol='tokens',outputCol='refined_tokens')
refined_df=stopword_removal.transform(tokenized_df)
refined_df.select(['user_id','tokens','refined_tokens']).show(10,False)

+-------+---------------------------------------------------+----------------------------------+
|user_id|tokens                                             |refined_tokens                    |
+-------+---------------------------------------------------+----------------------------------+
|1      |[i, really, liked, this, movie]                    |[really, liked, movie]            |
|2      |[i, would, recommend, this, movie, to, my, friends]|[recommend, movie, friends]       |
|3      |[movie, was, alright, but, acting, was, horrible]  |[movie, alright, acting, horrible]|
|4      |[i, am, never, watching, that, movie, ever, again] |[never, watching, movie, ever]    |
+-------+---------------------------------------------------+----------------------------------+

4

词袋模型

通过词袋可以数值化文本数据，以便将文本数据用于机器学习以及其它的任何分析。词袋允许我们将文本形式转换成数值向量形式，其中涉及单词在文本文档中出现的频率。通常有以下两种，一种是计数向量，一种是TF-IDF向量。

from pyspark.ml.feature import CountVectorizer
count_vec=CountVectorizer(inputCol='refined_tokens',outputCol='features')
cv_df=count_vec.fit(refined_df).transform(refined_df)
cv_df.select(['user_id','refined_tokens','features']).show(4,False)
count_vec.fit(refined_df).vocabulary
+-------+----------------------------------+--------------------------------+
|user_id|refined_tokens                    |features                        |
+-------+----------------------------------+--------------------------------+
|1      |[really, liked, movie]            |(11,[0,4,9],[1.0,1.0,1.0])      |
|2      |[recommend, movie, friends]       |(11,[0,6,10],[1.0,1.0,1.0])     |
|3      |[movie, alright, acting, horrible]|(11,[0,2,3,5],[1.0,1.0,1.0,1.0])|
|4      |[never, watching, movie, ever]    |(11,[0,1,7,8],[1.0,1.0,1.0,1.0])|
+-------+----------------------------------+--------------------------------+
['movie',
 'horrible',
 'really',
 'alright',
 'liked',
 'friends',
 'recommend',
 'never',
 'ever',
 'acting',
 'watching']

from pyspark.ml.feature import HashingTF,IDF
hashing_vec=HashingTF(inputCol='refined_tokens',outputCol='tf_features')
hashing_df=hashing_vec.transform(refined_df)
hashing_df.select(['user_id','refined_tokens','tf_features']).show(4,False)
+-------+----------------------------------+-------------------------------------------------------+
|user_id|refined_tokens                    |tf_features                                            |
+-------+----------------------------------+-------------------------------------------------------+
|1      |[really, liked, movie]            |(262144,[14,32675,155321],[1.0,1.0,1.0])               |
|2      |[recommend, movie, friends]       |(262144,[129613,155321,222394],[1.0,1.0,1.0])          |
|3      |[movie, alright, acting, horrible]|(262144,[80824,155321,236263,240286],[1.0,1.0,1.0,1.0])|
|4      |[never, watching, movie, ever]    |(262144,[63139,155321,203802,245806],[1.0,1.0,1.0,1.0])|
+-------+----------------------------------+-------------------------------------------------------+


tf_idf_vec=IDF(inputCol='tf_features',outputCol='tf_idf_features')
tf_idf_df=tf_idf_vec.fit(hashing_df).transform(hashing_df)
tf_idf_df.select(['user_id','tf_idf_features']).show(4,False)
+-------+----------------------------------------------------------------------------------------------------+
|user_id|tf_idf_features                                                                                     |
+-------+----------------------------------------------------------------------------------------------------+
|1      |(262144,[14,32675,155321],[0.9162907318741551,0.9162907318741551,0.0])                              |
|2      |(262144,[129613,155321,222394],[0.9162907318741551,0.0,0.9162907318741551])                         |
|3      |(262144,[80824,155321,236263,240286],[0.9162907318741551,0.0,0.9162907318741551,0.9162907318741551])|
|4      |(262144,[63139,155321,203802,245806],[0.9162907318741551,0.0,0.9162907318741551,0.9162907318741551])|
+-------+----------------------------------------------------------------------------------------------------+

5

文本分类

下面我们构建一个简单的文本分类模型，使用的数据及是开源的已标记的电影镜头评论数据，并且我们打算预测任意指定评论的情绪分类。整体流程如下程序所示。

from pyspark.sql.functions import rand 
from pyspark.sql.functions import length
from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType
from pyspark.sql.functions import *
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

text_df=spark.read.csv('Movie_reviews.csv',inferSchema=True,header=True,sep=',')

#创建标签列
text_df = text_df.withColumn("Label", text_df.Sentiment.cast('float')).drop('Sentiment')
#增加一个额外的length列来捕获评论的长度
text_df=text_df.withColumn('length',length(text_df['Review']))
#分词、去除停用词
tokenization=Tokenizer(inputCol='Review',outputCol='tokens')
tokenized_df=tokenization.transform(text_df)
stopword_removal=StopWordsRemover(inputCol='tokens',outputCol='refined_tokens')
refined_text_df=stopword_removal.transform(tokenized_df)


len_udf = udf(lambda s: len(s), IntegerType())
refined_text_df = refined_text_df.withColumn("token_count", len_udf(col('refined_tokens')))
refined_text_df.orderBy(rand()).show(10)
+--------------------+-----+------+--------------------+--------------------+-----------+
|              Review|Label|length|              tokens|      refined_tokens|token_count|
+--------------------+-----+------+--------------------+--------------------+-----------+
|da vinci code was...|  1.0|    37|[da, vinci, code,...|[da, vinci, code,...|          5|
|Not because I hat...|  0.0|    72|[not, because, i,...|[hate, harry, pot...|          6|
|I love Harry Potter.|  1.0|    20|[i, love, harry, ...|[love, harry, pot...|          3|
|and I love Da Vin...|  1.0|    71|[and, i, love, da...|[love, da, vinci,...|          7|
|Da Vinci Code = U...|  0.0|    72|[da, vinci, code,...|[da, vinci, code,...|         15|
|Brokeback Mountai...|  1.0|    34|[brokeback, mount...|[brokeback, mount...|          3|
|I think I hate Ha...|  0.0|    72|[i, think, i, hat...|[think, hate, har...|          9|
|Harry Potter is b...|  1.0|    26|[harry, potter, i...|[harry, potter, b...|          3|
|The Da Vinci Code...|  1.0|    30|[the, da, vinci, ...|[da, vinci, code,...|          4|
|Combining the opi...|  0.0|    71|[combining, the, ...|[combining, opini...|         10|
+--------------------+-----+------+--------------------+--------------------+-----------+

#将文本数据转换成数值特征
count_vec=CountVectorizer(inputCol='refined_tokens',outputCol='features')
cv_text_df=count_vec.fit(refined_text_df).transform(refined_text_df)

#select data for building model
model_text_df=cv_text_df.select(['features','token_count','Label'])

#创建输入特征
df_assembler = VectorAssembler(inputCols=['features','token_count'],outputCol='features_vec')
model_text_df = df_assembler.transform(model_text_df)

#划分训练、验证集
training_df,test_df=model_text_df.randomSplit([0.75,0.25])
#训练模型
log_reg=LogisticRegression(featuresCol='features_vec',labelCol='Label').fit(training_df)
#验证评估
results=log_reg.evaluate(test_df).predictions

#confusion matrix
true_postives = results[(results.Label == 1) & (results.prediction == 1)].count()
true_negatives = results[(results.Label == 0) & (results.prediction == 0)].count()
false_positives = results[(results.Label == 0) & (results.prediction == 1)].count()
false_negatives = results[(results.Label == 1) & (results.prediction == 0)].count()

recall = float(true_postives)/(true_postives + false_negatives)
print(recall)

precision = float(true_postives) / (true_postives + false_positives)
print(precision)

accuracy=float((true_postives+true_negatives) /(results.count()))
print(accuracy)