QCon北京「鸿蒙专场」火热来袭!即刻报名,与创新同行~ 了解详情
写点什么

使用 Spark Streaming 进行情感分析

  • 2016-05-24
  • 本文字数:2054 字

    阅读完需:约 7 分钟

这里将使用 Twitter 流式数据,它符合所有所需:持续而且无止境的数据源。

Spark Streaming

Spark Streaming 在电子书《手把手教你学习Spark》第六章有详细介绍,这里略过Streaming API 的详细介绍,直接进行程序开发 。

程序开发设置部分

程序开发起始部分需要做好准备工作。

复制代码
val config = new SparkConf().setAppName("twitter-stream-sentiment")
val sc = new SparkContext(config)
sc.setLogLevel("WARN")
val ssc = new StreamingContext(sc, Seconds(5))
System.setProperty("twitter4j.oauth.consumerKey", "consumerKey")
System.setProperty("twitter4j.oauth.consumerSecret", "consumerSecret")
System.setProperty("twitter4j.oauth.accessToken", accessToken)
System.setProperty("twitter4j.oauth.accessTokenSecret", "accessTokenSecret")
val stream = TwitterUtils.createStream(ssc, None)

这里创建一个 Spark Context sc,设置日志级别为 WARN 来消除 Spark 生成的日志。使用sc创建 Streaming Contextssc,然后设置 Twitter 证书来获得 Twitter 网站数据。

Twitter 上现在的趋势是什么?

很容易的能够找到任意给定时刻的 Twitter 趋势,仅仅需要计算数据流每个标签的数目。让我们看下 Spark 如何实现这个操作的。

复制代码
val tags = stream.flatMap { status =>
status.getHashtagEntities.map(_.getText)
}
tags.countByValue()
.foreachRDD { rdd =>
val now = org.joda.time.DateTime.now()
rdd
.sortBy(_._2)
.map(x => (x, now))
.saveAsTextFile(s"~/twitter/$now")
}

首先从 Tweets 获取标记,并计算标记的数量,按数量排序,然后持久化结果。我们基于前面的结果建立一个监控面板来跟踪趋势标签。作者的同事就可以创建一个广告标记(campaigns),并吸引更多的用户。

分析 Tweets

现在我们想增加一个功能来获得用户主要感兴趣的主题集。为了这个目的我们想对 Tweets 的大数据和食物两个不相关的主题进行情感分析。

有几种 API 可以在 Tweets 上做情感分析,但是作者选择斯坦福自然语言处理组开发的库来抽取相关情感。
build.sbt文件中增加相对应的依赖。

复制代码
libraryDependencies += "edu.stanford.nlp" % "stanford-corenlp" % "3.5.1"
libraryDependencies += "edu.stanford.nlp" % "stanford-corenlp" % "3.5.1" classifier "models"

现在,我们通过 Streaming 过滤一定的哈希标签,只选择感兴趣的 Tweets,如下所示:

复制代码
val tweets = stream.filter {t =>
val tags = t.getText.split(" ").filter(_.startsWith("#")).map(_.toLowerCase)
tags.contains("#bigdata") && tags.contains("#food")
}

得到 Tweets 上所有标签,然后标记出#bigdata 和 #food 两个标签。
接下来定一个函数从 Tweets 抽取相关的情感:

def detectSentiment(message: String): SENTIMENT_TYPE然后对 detectSentiment 进行测试以确保其可以工作:

复制代码
it("should detect not understood sentiment") {
detectSentiment("") should equal (NOT_UNDERSTOOD)
}
it("should detect a negative sentiment") {
detectSentiment("I am feeling very sad and frustrated.") should equal (NEGATIVE)
}
it("should detect a neutral sentiment") {
detectSentiment("I'm watching a movie") should equal (NEUTRAL)
}
it("should detect a positive sentiment") {
detectSentiment("It was a nice experience.") should equal (POSITIVE)
}
it("should detect a very positive sentiment") {
detectSentiment("It was a very nice experience.") should equal (VERY_POSITIVE)
}

完整列子如下:

复制代码
val data = tweets.map { status =>
val sentiment = SentimentAnalysisUtils.detectSentiment(status.getText)
val tags = status.getHashtagEntities.map(_.getText.toLowerCase)
(status.getText, sentiment.toString, tags)
}

data 中包含相关的情感。

和 SQL 协同进行分析

现在作者想把情感分析的数据存储在外部数据库,为了后续可以使用 SQL 查询。
具体操作如下:

复制代码
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
data.foreachRDD { rdd =>
rdd.toDF().registerTempTable("sentiments")
}

将 Dstream 转换成 DataFrame,然后注册成一个临时表,其他喜欢使用 SQL 的同事就可以使用不同的数据源啦。

sentiment 表可以被任意查询,也可以使用 Spark SQL 和其他数据源(比如,Cassandra 数据等)进行交叉查询。
查询 DataFrame 的列子:

sqlContext.sql("select * from sentiments").show()## 窗口操作

Spark Streaming 的窗口操作可以进行回溯数据,这在其他流式引擎中并没有。
为了使用窗口函数,你需要 checkpoint 流数据,具体详情见 http://spark.apache.org/docs/latest/streaming-programming-guide.html#checkpointing
简单的一个窗口操作:

复制代码
tags
.window(Minutes(1))
. (...)

结论

此列子虽然简单,但是其可以使用 Spark 解决实际问题。我们可以计算 Twitter 上主题趋势。

2016-05-24 17:455179
用户头像

发布了 43 篇内容, 共 29.4 次阅读, 收获喜欢 7 次。

关注

评论

发布
暂无评论
发现更多内容

独家好书丨《智算时代的容器技术演进与实践》免费下载

阿里巴巴云原生

阿里云 云原生 容器服务

【写作训练营打卡|10】内容创作方式

软件测试/测试管理圆桌讨论会 | 解锁测试管理的核心问题,提升您的管理实力!

霍格沃兹测试开发学社

测试开发 | 数据分析与智能训练:体育领域的技术革新

测吧(北京)科技有限公司

测试

火热的低代码,蕴藏怎样的机会和挑战?

互联网工科生

软件开发 低代码开发 JNPF

测试开发 | 数字化创新在文化保护中的崭新探索

测吧(北京)科技有限公司

测试

E往无前 | 腾讯云大数据ES日志轻接入和免运维最佳实践

腾讯云大数据

ES

产品经理该不该设计数据库表?

菜根老谭

每日一题:LeetCode-394. 字符串解码

Geek_4z9ami

Go 算法 递归 LeetCode 字符串

大模型训练:多模态预训练的未来

百度开发者中心

人工智能 大模型

详解ZooKeeper在微服务注册中心的应用

华为云开发者联盟

开源 开发 华为云 华为云开发者联盟

测试开发 | 人工智能的未来发展:科技新时代的奇迹之旅

测吧(北京)科技有限公司

测试

大模型之提示工程的使用与总结 主赛道:技术人的 2023 总结

Echo_Wish

技术 年度总结 大模型 2023 提示工程

智能引领决策:人工智能在决策支持系统中的崭新角色

测吧(北京)科技有限公司

测试

产品破局思考:一个思维,一种策略

菜根老谭

大模型训练中的Prompt Learning

百度开发者中心

人工智能 大模型

大模型训练:提高AI能力的重要策略

百度开发者中心

人工智能 大模型

测试开发 | 创造性领域中的计算机生成艺术

测吧(北京)科技有限公司

测试

测试开发 | 量子计算与人工智能的奇妙交融:开启全新计算时代

测吧(北京)科技有限公司

测试

软件测试/测试开发/人工智能 | 测试管理核心问题,提升管理实力!

测试人

人工智能 软件测试 测试开发

用AI为AI PC打造最强内核,英特尔是怎么做到的?

E科讯

我们为什么要标准定价?如何定价?

菜根老谭

产品经理 产品标准化

NineData成功举办《国产数据库共话未来趋势》技术沙龙:大咖云集,共话未来

NineData

postgresql 阿里云 国产数据库 polarDB NineData

大语言模型训练数据常见的4种处理方法

华为云开发者联盟

人工智能 华为云 华为云开发者联盟 大语言模型

Java基础知识:快速入门指南

技术冰糖葫芦

API

中手游全面启动鸿蒙原生游戏开发,为游戏产业发展注入新活力

新消费日报

使用Spark Streaming进行情感分析_语言 & 开发_侠天_InfoQ精选文章