介绍

自然语言处理是全球数据科学团队的重要过程之一。随着数据的不断增长&＃xff0c;大多数组织已经转移到大数据平台&＃xff0c;如apachehadoop和AWS、Azure和GCP等云产品。

这些平台不仅能够处理大数据&＃xff0c;使组织能够对非结构化数据(如文本分类)进行大规模分析。但在机器学习方面&＃xff0c;大数据系统和机器学习工具之间仍然存在差距。

流行的机器学习python库&＃xff0c;如scikit-learn和Gensim&＃xff0c;经过高度优化&＃xff0c;可以在单节点计算机上执行&＃xff0c;而不是为分布式环境设计的。

Apache Spark MLlib是许多帮助弥合这一差距的工具之一&＃xff0c;它提供了大多数机器学习模型&＃xff0c;如线性回归、Logistic回归、支持向量机、随机森林、K-means、LDA等&＃xff0c;以执行最常见的机器学习任务。

除了机器学习算法&＃xff0c;Spark MLlib还提供了大量的特征变换器&＃xff0c;如Tokenizer、StopWordRemover、n-grams和countvector、TF-IDF和Word2Vec等。

虽然这些转换器和提取器足以构建基本的NLP管道&＃xff0c;但是要构建一个更全面和生产级的管道&＃xff0c;我们需要更先进的技术&＃xff0c;如词干分析、词法化、词性标记和命名实体识别。

Spark NLP提供了各种注释器来执行高级NLP任务。有关更多信息&＃xff0c;请在网站上查看注释器列表及其用法

https://nlp.johnsnowlabs.com/docs/en/annotators。

设置环境

让我们继续看看如何在AWS EMR上设置Spark NLP。

1.在启动EMR集群之前&＃xff0c;我们需要创建一个引导操作。引导操作用于设置其他软件或自定义群集节点的配置。以下是可用于在EMR集群上设置Spark NLP的引导操作&＃xff0c;

#!/bin/bashsudo yum install -y python36-devel python36-pip python36-setuptools python36-virtualenvsudo python36 -m pip install --upgrade pip#sudo python36 -m pip install pandas#sudo python36 -m pip install boto3#sudo python36 -m pip install re#sudo python36 -m pip install spark-nlp&＃61;&＃61;2.4.5

创建shell脚本之后&＃xff0c;将该脚本复制到AWS S3中的一个位置。你还可以根据需要安装其他python包。

2.我们可以使用AWS控制台、API或python中的boto3库来启动EMR集群。使用Python的好处是&＃xff0c;无论何时需要实例化集群或将其添加到工作流中&＃xff0c;都可以重用代码。

下面是实例化EMR集群的python代码。

import boto3region_name&＃61;&＃39;region_name&＃39;def get_security_group_id(group_name, region_name): ec2 &＃61; boto3.client(&＃39;ec2&＃39;, region_name&＃61;region_name) response &＃61; ec2.describe_security_groups(GroupNames&＃61;[group_name]) return response[&＃39;SecurityGroups&＃39;][0][&＃39;GroupId&＃39;]emr &＃61; boto3.client(&＃39;emr&＃39;, region_name&＃61;region_name)cluster_response &＃61; emr.run_job_flow( Name&＃61;&＃39;cluster_name&＃39;, # 更新值 ReleaseLabel&＃61;&＃39;emr-5.27.0&＃39;, LogUri&＃61;&＃39;s3_path_for_logs&＃39;, # 更新值 Instances&＃61;{ &＃39;InstanceGroups&＃39;: [ { &＃39;Name&＃39;: "Master nodes", &＃39;Market&＃39;: &＃39;ON_DEMAND&＃39;, &＃39;InstanceRole&＃39;: &＃39;MASTER&＃39;, &＃39;InstanceType&＃39;: &＃39;m5.2xlarge&＃39;, # 根据要求进行变更 &＃39;InstanceCount&＃39;: 1 #对于主节点高可用性&＃xff0c;设置计数大于1 }, { &＃39;Name&＃39;: "Slave nodes", &＃39;Market&＃39;: &＃39;ON_DEMAND&＃39;, &＃39;InstanceRole&＃39;: &＃39;CORE&＃39;, &＃39;InstanceType&＃39;: &＃39;m5.2xlarge&＃39;, # 根据要求进行变更 &＃39;InstanceCount&＃39;: 2 } ], &＃39;KeepJobFlowAliveWhenNoSteps&＃39;: True, &＃39;Ec2KeyName&＃39; : &＃39;key_pair_name&＃39;, # 更新值 &＃39;EmrManagedMasterSecurityGroup&＃39;: get_security_group_id(&＃39;ElasticMapReduce-master&＃39;, region_name&＃61;region_name) &＃39;EmrManagedSlaveSecurityGroup&＃39;: get_security_group_id(&＃39;ElasticMapReduce-master&＃39;, region_name&＃61;region_name) }, BootstrapActions&＃61;[ { &＃39;Name&＃39;:&＃39;install_dependencies&＃39;, &＃39;ScriptBootstrapAction&＃39;:{ &＃39;Args&＃39;:[], &＃39;Path&＃39;:&＃39;path_to_bootstrapaction_on_s3&＃39; # 更新值 } }], Steps &＃61; [], VisibleToAllUsers&＃61;True, JobFlowRole&＃61;&＃39;EMR_EC2_DefaultRole&＃39;, ServiceRole&＃61;&＃39;EMR_DefaultRole&＃39;, Applications&＃61;[ { &＃39;Name&＃39;: &＃39;hadoop&＃39; }, { &＃39;Name&＃39;: &＃39;spark&＃39; }, { &＃39;Name&＃39;: &＃39;hive&＃39; }, { &＃39;Name&＃39;: &＃39;zeppelin&＃39; }, { &＃39;Name&＃39;: &＃39;presto&＃39; } ], Configurations&＃61;[ # YARN { "Classification": "yarn-site", "Properties": {"yarn.nodemanager.vmem-pmem-ratio": "4", "yarn.nodemanager.pmem-check-enabled": "false", "yarn.nodemanager.vmem-check-enabled": "false"} }, # HADOOP { "Classification": "hadoop-env", "Configurations": [ { "Classification": "export", "Configurations": [], "Properties": {"JAVA_HOME": "/usr/lib/jvm/java-1.8.0"} } ], "Properties": {} }, # SPARK { "Classification": "spark-env", "Configurations": [ { "Classification": "export", "Configurations": [], "Properties": {"PYSPARK_PYTHON":"/usr/bin/python3", "JAVA_HOME": "/usr/lib/jvm/java-1.8.0"} } ], "Properties": {} }, { "Classification": "spark", "Properties": {"maximizeResourceAllocation": "true"}, "Configurations": [] }, { "Classification": "spark-defaults", "Properties": { "spark.dynamicAllocation.enabled": "true" #default is also true } } ] )

注意&＃xff1a;请确保你对用于日志记录和存储引导操作脚本的S3 bucket具有正确的访问权限。

基于Spark-NLP的BBC文章文本分类

现在我们已经准备好集群了&＃xff0c;让我们使用Spark NLP和Spark MLlib在BBC数据上构建一个简单的文本分类示例。

1.初始化Spark

我们将导入所需的库并使用不同的配置参数初始化spark会话。配置值取决于我的本地环境。相应地调整参数。

# 导入Spark NLPfrom sparknlp.base import *from sparknlp.annotator import *from sparknlp.pretrained import PretrainedPipelineimport sparknlpfrom pyspark.sql import SparkSessionfrom pyspark.ml import Pipeline# 使用Spark NLP启动Spark会话#spark &＃61; sparknlp.start()spark &＃61; SparkSession.builder .appName("BBC Text Categorization") .config("spark.driver.memory","8G") change accordingly .config("spark.memory.offHeap.enabled",True) .config("spark.memory.offHeap.size","8G") .config("spark.driver.maxResultSize", "2G") .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.11:2.4.5") .config("spark.kryoserializer.buffer.max", "1000M") .config("spark.network.timeout","3600s") .getOrCreate()

2.加载文本数据

我们将使用BBC的数据。你可以从这个链接下载数据。下载以下数据后&＃xff0c;使用spark代码加载&＃xff1b;

https://www.kaggle.com/yufengdev/bbc-text-categorization?#Get-the-data

# 文件位置和类型file_location &＃61; r&＃39;pathobbc-text.csv&＃39;file_type &＃61; "csv"# CSVinfer_schema &＃61; "true"first_row_is_header &＃61; "true"delimiter &＃61; ","df &＃61; spark.read.format(file_type) .option("inferSchema", infer_schema) .option("header", first_row_is_header) .option("sep", delimiter) .load(file_location)df.count()

3.将数据集拆分为训练集和测试集

与python使用scikit learn分割数据不同&＃xff0c;Spark Dataframe有一个内置函数randomSplit()来执行相同的操作。

(trainingData, testData) &＃61; df.randomSplit([0.7, 0.3], seed &＃61; 100)

randomSplit()函数需要两个参数viz。权重数组和seed。在我们的例子中&＃xff0c;我们将使用70/30分割&＃xff0c;其中70%是训练数据&＃xff0c;30%是测试数据。

4.使用Spark NLP的NLP管道

让我们继续使用Spark NLP构建NLP管道。Spark NLP最大的优点之一是它与Spark MLLib模块本机集成&＃xff0c;有助于构建由transformers和estimators组成的综合ML管道。

这个管道可以包括诸如CountVectorizer或HashingTF和IDF之类的特征提取模块。我们还可以在这个管道中包含一个机器学习模型。

下面是由具有特征提取和机器学习模型的NLP管道组成的示例&＃xff1b;

from pyspark.ml.feature import HashingTF, IDF, StringIndexer, SQLTransformer,IndexToStringfrom pyspark.ml.classification import LogisticRegressionfrom pyspark.ml.evaluation import MulticlassClassificationEvaluator# 转换text列为nlp文件document_assembler &＃61; DocumentAssembler() .setInputCol("text") .setOutputCol("document")#将文档转换为标识数组tokenizer &＃61; Tokenizer() .setInputCols(["document"]) .setOutputCol("token")# 清理标识normalizer &＃61; Normalizer() .setInputCols(["token"]) .setOutputCol("normalized")# 删除停用词stopwords_cleaner &＃61; StopWordsCleaner() .setInputCols("normalized") .setOutputCol("cleanTokens") .setCaseSensitive(False)stemmer &＃61; Stemmer() .setInputCols(["cleanTokens"]) .setOutputCol("stem")# 将自定义文档结构转换为标识数组。finisher &＃61; Finisher() .setInputCols(["stem"]) .setOutputCols(["token_features"]) .setOutputAsArray(True) .setCleanAnnotations(False)# 生成频率hashingTF &＃61; HashingTF(inputCol&＃61;"token_features", outputCol&＃61;"rawFeatures", numFeatures&＃61;1000)# 生成逆文档频率idf &＃61; IDF(inputCol&＃61;"rawFeatures", outputCol&＃61;"features", minDocFreq&＃61;5)# 将标签(字符串)转换为整数。label_stringIdx &＃61; StringIndexer(inputCol &＃61; "category", outputCol &＃61; "label")# 定义一个简单的多项式逻辑回归模型。尝试不同的超参数组合&＃xff0c;看看哪个更适合你的数据。你也可以尝试不同的算法来比较分数。lr &＃61; LogisticRegression(maxIter&＃61;10, regParam&＃61;0.3, elasticNetParam&＃61;0.0)# 将索引(整数)转换为相应的类标签label_to_stringIdx &＃61; IndexToString(inputCol&＃61;"label", outputCol&＃61;"article_class")# 定义nlp管道nlp_pipeline &＃61; Pipeline( stages&＃61;[document_assembler, tokenizer, normalizer, stopwords_cleaner, stemmer, finisher, hashingTF, idf, label_stringIdx, lr, label_to_stringIdx])

5.训练模型

现在我们的NLP管道已经准备好了&＃xff0c;让我们根据训练数据训练我们的模型。

# 在训练数据上拟合管道pipeline_model &＃61; nlp_pipeline.fit(trainingData)

6.执行预测

一旦训练完成&＃xff0c;我们就可以预测测试数据上的类标签。

# 对测试数据进行预测predictions &＃61; pipeline_model.transform(testData)

7. 评估模型

对训练后的模型进行评估对于理解模型如何在看不见的数据上运行是非常重要的。我们将看到3个流行的评估指标&＃xff0c;准确度、精确度和召回率。

1. 准确度

# 导入evaluatorfrom pyspark.ml.evaluation import MulticlassClassificationEvaluatorevaluator &＃61; MulticlassClassificationEvaluator( labelCol&＃61;"label", predictionCol&＃61;"prediction", metricName&＃61;"accuracy")accuracy &＃61; evaluator.evaluate(predictions)print("Accuracy &＃61; %g" % (accuracy))print("Test Error &＃61; %g " % (1.0 - accuracy))

1. 精确度

evaluator &＃61; MulticlassClassificationEvaluator( labelCol&＃61;"label", predictionCol&＃61;"prediction", metricName&＃61;"weightedPrecision")accuracy &＃61; evaluator.evaluate(predictions)print("Accuracy &＃61; %g" % (accuracy))print("Test Error &＃61; %g " % (1.0 - accuracy))

1. 召回率

evaluator &＃61; MulticlassClassificationEvaluator( labelCol&＃61;"label", predictionCol&＃61;"prediction", metricName&＃61;"weightedRecall")accuracy &＃61; evaluator.evaluate(predictions)print("Accuracy &＃61; %g" % (accuracy))print("Test Error &＃61; %g " % (1.0 - accuracy))

根据业务用例&＃xff0c;你可以决定使用哪个度量来评估模型。

例如.如果一个机器学习模型被设计用来根据某些参数来检测癌症&＃xff0c;那么最好使用召回率&＃xff0c;因为公司无法承受假负例(一个患有癌症但模型没有检测到癌症的人)&＃xff0c;而如果机器学习模型旨在生成用户推荐&＃xff0c;公司可以负担得起误报(10条建议中有8条符合用户配置文件)&＃xff0c;因此可以使用精确度作为评估指标。

8. 保存管道模型

在成功地训练、测试和评估模型之后&＃xff0c;你可以将模型保存到磁盘&＃xff0c;并在不同的Spark应用程序中使用它。要将模型保存到光盘&＃xff0c;请使用以下代码&＃xff1b;

pipeline_model.save(&＃39;/path/to/storage_location&＃39;)

结论

Spark NLP提供了大量的注释器和转换器来构建数据预处理管道。Sparl NLP与Spark MLLib无缝集成&＃xff0c;使我们能够在分布式环境中构建端到端的自然语言处理项目。

在本文中&＃xff0c;我们研究了如何在AWS EMR上安装Spark NLP并实现了BBC数据的文本分类。我们还研究了Spark MLlib中的不同评估指标&＃xff0c;并了解了如何存储模型以供进一步使用。