在本教程中，您将学习如何使用积极的情绪分数来丰富COVID19推文数据。您将利用PySpark和认知服务并学习增强分析。
什么是增强分析？
根据Gartner的报告，增强分析是使用诸如机器学习和AI之类的技术来辅助数据准备，见解生成。它的主要目标是帮助更多的人从数据中获得价值，并以一种简单的对话方式获得见解。在我们的示例中，我们从一条推文中提取了积极的情绪评分，以帮助您了解对COVID-19的总体情绪。
什么是PySpark？ 福州小程序开发
PySpark是我们用于Apache Spark和Python的框架。在此处了解更多信息。
什么是情感分析？
情感分析是自然语言处理（NLP）的一部分-自然语言处理使用结合了文本分析，计算语言学等功能，可以更系统地研究情感状态和主观信息，例如推文。在我们的示例中，我们将看到如何从COVID-19推文文本中提取积极的情绪评分。在本教程中，您将利用Azure认知服务，它为我们提供了情感分析功能。使用它时，我们可以利用TextAnalyticsClient 客户端库或REST API。今天，您将使用REST API，因为它为我们提供了更大的灵活性。
先决条件
- 带有笔记本的Apache Spark环境，可以是Databricks，也可以通过运行以下命令通过docker启动本地环境：
docker run -it -p 8888:8888 jupyter/pyspark-notebook - Azure免费帐户
- 下载Kaggle COVID-19鸣叫数据
- 认知服务免费帐户（查看下面的图片）
分步教程-完整数据管道：
在本分步教程中，您将学习如何使用PySpark加载数据，创建用户定义函数以连接到Sentiment Analytics API，添加情感数据并将所有内容保存到Parquet格式文件。
现在，您需要将上载数据提取到Apache Spark环境中，而不是Databricks或PySpark jupyter笔记本。对于Databricks使用此功能，对于juypter使用此功能。
对于这两种情况，您都需要file_location = "/FileStore/tables/covid19_tweets.csv"确保对其进行记录。
用PySpark加载数据
这是将数据加载到PySpark DataFrame对象的方式，spark将尝试直接从CSV推断架构。您会注意到的一件事是，在使用CSV并推断架构时，Spark通常将大多数列称为String格式。
inputDF =火花。阅读。\
格式（“ com.databricks.spark.csv”）。\
选项（“ header”，“ true”）。\
选项（“ inferSchema”，“ true”）。加载（“ /FileStore/tables/covid19_tweets.csv”）
为我们的数据提供更准确的架构：
在此处，您定义，expectedSchema然后强制转换数据以使其匹配。您将使用StructType 和StructField哪些Spark SQL数据类型来帮助您定义架构。
withColumn功能会根据您为其提供的名称和值创建一个具有所需列的新DataFrame。
来自 pyspark。sql。输入类型 *
来自 pyspark。sql。功能 导入 *
＃创建期望的模式
expectedSchema = StructType（[
StructField（“ user_name”，StringType（），True），
StructField（“ user_location”，StringType（），True），
StructField（“ user_description”，StringType（），True），
StructField（“ user_created”，StringType（），True），
StructField（“ user_followers”，FloatType（），True），
StructField（“ user_friends”，FloatType（），True），
StructField（“ user_favourites”，FloatType（），True），
StructField（“ user_verified”，BooleanType（），True），
StructField（“ date”，StringType（），True），
StructField（“ text”，StringType（），True），
StructField（“ hashtags ”，StringType（），True），
StructField（“ source”，StringType（），True），
StructField（“ is_retweet”，BooleanType（），True）
]）
现在，让我们使用正确的架构创建新的DataFrame！
请注意，您将新模式分配给inputDF，这意味着您将不再有权访问旧的DataFrame。
＃设置数据类型-将数据强制转换为与模式匹配的列
inputDF = inputDF \
。withColumn（“ user_name”，inputDF [ “ user_name” ]。cast（“ string”））\
。withColumn（“ user_location”，inputDF [ “ user_location” ]。cast（“ string”））\
。withColumn（“ user_description”，inputDF [ “ user_description” ]。cast（“字符串”））\
。withColumn（“ user_created”，inputDF [ “ user_created” ]。cast（“ string”））\
。withColumn（“ user_followers”，inputDF [ “ user_followers” ]。cast（“ float”））\
。withColumn（“ user_friends”，inputDF [ “ user_friends” ]。cast（“ float”））\
。withColumn（“ user_favourites”，inputDF [ “ user_favourites” ]。cast（“ float”））\
。withColumn（“ user_verified”，inputDF [ “ user_verified” ]。cast（“ boolean”））\
。withColumn（“ date”，inputDF [ “ date” ]。cast（“ string”））\
。withColumn（“ text”，inputDF [ “ text” ]。cast（“ string”））\
。withColumn（“ hashtags ”，inputDF [ “ hashtags ” ]。cast（“ string”））\
。withColumn（“ source”，inputDF [ “ source” ]。cast（“ string”））\
。withColumn（“ is_retweet”，inputDF [ “ is_retweet” ]。cast（“ boolean”））\
使用REST API连接到情绪分析
为了连接和使用情感分析服务，我们需要提供情感分析端点和访问密钥。两者都可以在中找到。
查找端点，可以从“概述”部分或“键和端点”中找到。
查找访问密钥：
找到密钥和端点后，对于生产和团队合作，您需要将它们存储在安全的地方，尝试在代码中的自由文本中提供保存密钥，这是不安全的。最终，黑客可能会在您的云环境中挖掘比特币。
对于Databricks，您可以利用dbutils.secrets功能。这是设置方法。
如果您在本地使用juypter PySpark笔记本，则可以使用纯文本格式，但是将代码提交到git repo时请记住将其删除。
这是使用dbutils的方法，并为其提供了范围和键名。
在此代码段中，作用域名为-mle2ebigdatakv，键的名称为sentimentEndpoint 和sentimentAccessKeys。
＃提供端点和密钥
sentimentEndpoint = dbutils。机密。get（scope = “ mle2ebigdatakv”，key = “ sentimentEndpoint”）
sentimentAccessKeys = dbutils。机密。get（scope = “ mle2ebigdatakv”，key = “ sentimentAccessKeys”）
让我们构建连接本身，情感分析希望收到一个像对象一样的文档，因为您将使用python字典并构建具有ID的doc请求。每个请求的ID必须唯一。
注意这里的language_api_url变量，这是您构造认知分析请求的地方，要求使用3.0版的文本分析情感。
汇入 要求
＃使用language_api_url构建其余的API请求
language_api_url =情感 端点+ “ /text/analytics/v3.0/sentiment”
标头= { “ Ocp-Apim-Subscription-Key”：sentimentAccessKeys }
def constractDocRequest（文字）：
docRequest = {}
doc = {}
doc [ “ id” ] =文字
doc [ “ text” ] =文字
docRequest [ “ documents” ] = [ doc ]
返回 docRequest
尝试使用一些文本来运行它，您将看到响应与分数情感一致positive, netural且为负。
这是响应的结构：
{
“文件”：[
{
“ id”：“ 1”，
“ sentiment”：“ positive”，
“ confidenceScores”：{
“正数”：1.0，
“中立”：0.0，
“负数”：0.0
}，
“句子”：[
{
“ sentiment”：“ positive”，
“ confidenceScores”：{
“正数”：1.0，
“中立”：0.0，
“负数”：0.0
}，
“偏移”：0，
“长度”：66，
“ text”：“ covid19一点也不吓人，实际上并不是兴旺的机会”
}
]，
“警告”：[]
}
]，
“错误”：[]，
“ modelVersion”：“ 2020-04-01”
}
让我们构建一个python功能，以提取情感并通过UDF（用户定义函数）api向PySpark注册该函数。
您需要确保实际上是从REST API取回文档，并且还需要保护您的功能免于向情感分析服务发送空文本，因为它将解决错误。
这是将所有内容连接在一起的方式：
来自 pyspark。sql。函数 导入 udf
＃从返回的json文档中提取情感
def extractSentiment（doc，sentimentType）：
如果 DOC == {}或 没有 '文件' 的 文档：
返回 0.0
返回 浮点数（doc [ 'documents' ] [ 0 ] [ 'confidenceScores' ] [ sentimentType ]）
＃提取积极情绪的功能
def getPositiveSentiment（text）：
如果 布尔（文本。条（））==假：
返回 0.0
正=随机。均匀的（0，1）
返回 正数
＃创建udf函数指针
get_positive_sentiment = UDF（getPositiveSentiment，StringType（））
＃使用新的列创建一个新的DF，以重新设置积极的情绪评分
enrichedDF_positiveSentiment = inputDF。withColumn（'positive_sentiment'，get_positive_sentiment（inputStream [ “ text” ]））
充实数据之后，将其保存到存储中以备将来使用是很重要的，您将其保存为镶木地板格式，这将保持架构完整。与基于行的文件（例如CSV）相比，Apache Parquet格式旨在提供高效的列式数据存储，因为它可以提供更好的压缩效果，并在以后更快地扫描列的子集。相对于CSV，我们必须读取整个文件和列以仅查询其中的一部分。
这是通过PySpark完成的：
＃将处理后的数据流式传输到木地板中，以供数据科学探索和构建ML模型
richedDF_poisitveSentiment。写 \
。格式（“实木复合地板”）\
。保存（“ / DATALAKE / COVID19_TWEETS / REFINED / WITH_SENTIMENT /”）
您可以决定数据的名称和结构，但要确保指出，该数据现在包含一个带有积极情绪的新列。
以后可以在各种数据可视化工具中使用此数据或供研究人员使用。
可能发生的问题
{ 'error'：{ 'code'：'InvalidRequest'，
'innererror'：{ 'code'：'EmptyRequest'， 'message'：'Request body必须存在。'
}，
'message'：'无效请求。}
}
请注意该错误消息，当您看到这种错误时，可能是因为您对Cognitive Services失去了报价。如果您了解该服务并进行了尝试，最好使用一些示例而不是整个数据集，因为您可能会在免费层中用尽配额，因为它可以进行多达5千次交易。
而已！
本教程向您介绍了如何利用现有的REST API服务来丰富您的数据，以用于将来的工作和增强的分析。
要了解更多信息，请查看GitHub repo。很高兴回答您的问题，并在Twitter上与您联系。
