七、大数据和 Twitter 情感分析

“数据是新的石油。”

————未知

在本章中，我们将研究 AI 和数据科学的两个重要领域：自然语言处理（NLP）和大数据分析。对于支持的示例应用，我们重新实现了 Twitter 主题标签项目的情感分析，该项目在第 1 章“开发人员对数据科学的观点”中进行了介绍，我们利用 Jupyter 笔记本和 PixieDust 构建实时的仪表板，以分析从相关的推文流到特定实体（例如公司提供的产品）的数据，以提供情感信息，以及有关从相同推文中提取的趋势实体的其他信息。在本章的最后，读者将学习如何将基于云的 NLP 服务（例如，IBM Watson 自然语言理解）集成到其应用中，以及如何在（Twitter）规模上使用诸如 Apache Spark。

与往常一样，我们将展示如何通过将实时仪表板实现为直接在 Jupyter 笔记本中运行的 PixieApp 来实现分析的操作。

Apache Spark 入门

大数据一词可能会感到模糊不清。考虑任何数据集大数据的截止点是什么？是 10 GB，100 GB，1 TB 还是更多？我喜欢的一个定义是：大数据是指数据无法容纳在单台计算机上可用的内存中。多年以来，数据科学家一直被迫对大型数据集进行采样，因此它们可以放入一台机器中，但是随着并行计算框架能够将数据分布到一组机器中，并行处理框架开始发生变化，这种情况就开始发生变化。整个数据集，当然前提是集群具有足够的计算机。同时，随着云技术的进步，可以按需提供适应数据集大小的机器集群。

如今，有多个框架（大多数时候可以作为开源使用），可以提供强大，灵活的并行计算功能。最受欢迎的包括 Apache Hadoop，Apache Spark 和 Dask。对于我们的 Twitter 情感分析应用，我们将使用 Apache Spark，它在可伸缩性，可编程性和速度方面提供出色的性能。此外，许多云提供商提供了一些 Spark 即服务的功能，使您能够在几分钟内按需创建大小合适的 Spark 集群。

一些 Spark 即服务的云提供商包括：

Microsoft Azure
亚马逊网络服务
Google 云端
Databricks
IBM Cloud

注意

注意：Apache Spark 也可以很容易地安装在本地计算机上以进行测试，在这种情况下，群集节点是使用线程模拟的。

Apache Spark 架构

以下图显示了 Apache Spark 框架的主要组件：

Spark 高级架构

Spark SQL：此组件的核心数据结构是 Spark DataFrame，它使了解 SQL 语言的用户可以轻松处理结构化数据。
Spark Streaming：模块用于处理流数据。稍后我们将看到，我们将在示例应用中使用此模块，尤其是使用结构化流（在 Spark 2.0 中引入）。
MLlib：模块，提供了功能丰富的机器学习库，可在 Spark 规模上工作。
GraphX：模块用于执行图并行计算。

主要有两种使用 Spark 集群的方法，如下图所示：

使用 Spark 集群的两种方法

spark-submit：用于在集群上启动 Spark 应用的 Shell 脚本
笔记本：以交互方式针对 Spark 集群执行代码语句

关于spark-submit shell 脚本的内容不在本书的讨论范围内，但是可以在以下位置找到官方文档。在本章的其余部分，我们将重点介绍通过 Jupyter 笔记本与 Spark 集群进行交互。

配置笔记本来配合 Spark

本节中的说明仅涉及在本地安装 Spark 以进行开发和测试。在群集中手动安装 Spark 超出了本书的范围。如果需要一个真正的集群，强烈建议使用基于云的服务。

默认情况下，本地 Jupyter 笔记本安装了纯 Python 内核。要使用 Spark，用户必须使用以下步骤：

通过从这个页面下载二进制发行版在本地安装 Spark。
使用以下命令在临时目录中生成内核规范：
```
ipython kernel install --prefix /tmp
```
注意

注意：只要声明以下消息，上述命令可能会生成警告消息，可以安全地忽略该警告消息：

Installed kernelspec python3 in /tmp/share/jupyter/kernels/python3

转到/tmp/share/jupyter/kernels/python3，然后编辑kernel.json文件，将以下键添加到 JSON 对象（将<<spark_root_path>>替换为安装 Spark 的目录路径，将<<py4j_version>>替换为系统上安装的版本）：

"env": {
    "PYTHONPATH": "<<spark_root_path>>/python/:<<spark_root_path>>/python/lib/py4j-<<py4j_version>>-src.zip",
    "SPARK_HOME": "<<spark_root_path>>",
    "PYSPARK_SUBMIT_ARGS": "--master local[10] pyspark-shell",
    "SPARK_DRIVER_MEMORY": "10G",
    "SPARK_LOCAL_IP": "127.0.0.1",
    "PYTHONSTARTUP": "<<spark_root_path>>/python/pyspark/shell.py"
}

您可能还需要自定义display_name键，使其具有唯一性，并且可以从 Juptyer UI 轻松识别。如果需要了解现有内核的列表，可以使用以下命令：
```
jupyter kernelspec list
```
前面的命令将为您提供内核名称列表以及本地文件系统上的关联路径。从路径中，您可以打开kernel.json文件来访问display_name值。例如：
```
 Available kernels:
 pixiedustspark16
 /Users/dtaieb/Library/Jupyter/kernels/pixiedustspark16
 pixiedustspark21
 /Users/dtaieb/Library/Jupyter/kernels/pixiedustspark21
 pixiedustspark22
 /Users/dtaieb/Library/Jupyter/kernels/pixiedustspark22
 pixiedustspark23
 /Users/dtaieb/Library/Jupyter/kernels/pixiedustspark23
```
使用以下命令将内核与已编辑的文件一起安装：
```
jupyter kernelspec install /tmp/share/jupyter/kernels/python3
```
注意

注意：根据环境，运行前面的命令时，您可能会收到“权限被拒绝”错误。在这种情况下，您可能希望使用sudo以管理员权限运行命令或使用--user开关，如下所示：

jupyter kernelspec install --user /tmp/share/jupyter/kernels/python3

有关安装选项的更多信息，可以使用-h开关。例如：
```
 jupyter kernelspec install -h
```
重新启动笔记本服务器并开始使用新的 PySpark 内核。

幸运的是，PixieDust 提供了一个install脚本来自动执行上述手动步骤。

注意

您可以在此处找到有关此脚本的详细文档。

简而言之，使用自动 PixieDust install脚本需要发出以下命令并遵循屏幕上的说明：

jupyter pixiedust install

我们将在本章稍后深入研究 Spark 编程模型，但是现在，让我们在下一部分中定义 Twitter 情感分析应用的 MVP 要求。

Twitter 情感分析应用

与往常一样，我们首先定义 MVP 版本的要求：

连接到 Twitter 以获取由用户提供的查询字符串过滤的实时推文流
丰富推文以添加情感信息和从文本中提取的相关实体
使用实时图表显示仪表板，其中包含有关数据的各种统计信息，并按指定的时间间隔进行更新
系统应该能够扩展到 Twitter 数据大小

以下图显示了我们的应用架构的第一个版本：

Twitter 情感架构版本 1

对于版本 1，该应用将完全在单个 Python 笔记本中实现，并将调出 NLP 部分的外部服务。为了进行扩展，我们当然必须将笔记本外部的某些处理外部化，但是为了进行开发和测试，我发现能够在单个笔记本中包含整个应用可以显着提高生产力。

至于库和框架，我们将使用 Tweepy 将连接到 Twitter，Apache Spark 结构化流，用于处理分布式集群和 Watson Developer Cloud Python SDK 来访问 IBM Watson Natural 语言理解服务。

第 1 部分——使用 Spark 结构化流获取数据

为了获取数据，我们使用 Tweepy，它提供了一个优雅的 Python 客户端库来访问 Twitter API。 Tweepy 涵盖的 API 非常广泛，详细介绍超出了本书的范围，但是您可以在 Tweepy 官方网站上找到完整的 API 参考。

您可以使用pip install命令直接从 PyPi 安装 Tweepy 库。以下命令显示如何使用!指令从笔记本计算机安装它：

!pip install tweepy

注意

注意：当前使用的 Tweepy 版本是 3.6.0。安装库后，不要忘记重新启动内核。

数据管道的架构图

在我们开始深入研究数据管道的每个组件之前，最好先了解其总体架构并了解计算流程。

如下图所示，我们首先创建一个 Tweepy 流，该流将原始数据写入 CSV 文件。然后，我们创建一个 Spark Streaming DataFrame，该框架读取 CSV 文件，并定期使用新数据进行更新。从 Spark StreamingDataFrame中，我们使用 SQL 创建一个 Spark 结构化查询并将其结果存储在 Parquet 数据库中：

流计算流程

通过 Twitter 执行认证

在使用任何 Twitter API 之前，建议先通过系统进行认证。 OAuth 2.0 协议是最常用的认证机制之一，它使第三方应用能够访问网络上的服务。您需要做的第一件事是获取 OAuth 协议用来验证您身份的一组密钥字符串：

用户密钥：唯一标识客户端应用的字符串（也称为 API 密钥）。
使用者密码：仅应用和 Twitter OAuth 服务器知道的密码字符串。可以认为它就像一个密码。
访问令牌：使用字符串来验证您的请求。在授权阶段还可以使用此令牌来确定应用的访问级别。
访问令牌密钥：类似于用户密钥，这是与访问令牌一起发送的秘密字符串，用作密码。

要生成上述密钥字符串，您需要转到这个页面，使用常规的 Twitter 用户 ID 和密码进行认证，然后按照以下步骤操作：

使用创建新应用按钮创建一个新的 Twitter 应用。
填写应用详细信息，同意开发者协议，然后单击创建您的 Twitter 应用按钮。

提示

注意：请确保您的手机号码已添加到个人资料中，否则在创建 Twitter 应用时会出现错误。

您可以为 必填网站 的强制输入提供随机 URL，并将 URL 的输入保留为空白，因为这是可选的回调 URL。
单击密钥和访问令牌选项卡以获取使用者和访问令牌。您可以随时使用此页面上的按钮重新生成这些令牌。如果这样做，则还需要更新您的应用代码中的值。

为了更轻松地维护代码，让我们将这些标记放在笔记本顶部的自己的变量中，并创建tweepy.OAuthHandler类，稍后我们将使用它：

from tweepy import OAuthHandler
# Go to http://apps.twitter.com and create an app.
# The consumer key and secret will be generated for you after
consumer_key="XXXX"
consumer_secret="XXXX"

# After the step above, you will be redirected to your app's page.
# Create an access token under the "Your access token" section
access_token="XXXX"
access_token_secret="XXXX"

auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)

创建 Twitter 流

为了实现我们的应用，我们只需要使用此处记录的 Twitter 流 API。在此步骤中，我们创建一个 Twitter 流，将输入的数据存储到本地文件系统上的 CSV 文件中。这是使用从tweepy.streaming.StreamListener继承的自定义RawTweetsListener类完成的。通过覆盖on_data方法，可以完成对传入数据的自定义处理。

在我们的例子中，我们想使用来自标准 Python csv模块的DictWriter将传入的数据从 JSON 转换为 CSV。由于 Spark Streaming 文件输入源仅在输入目录中创建新文件时才触发，因此我们不能简单地将数据追加到现有文件中。取而代之的是，我们将数据缓冲到一个数组中，并在缓冲区达到容量后将其写入磁盘。

注意

为简单起见，该实现不包括在处理完文件后清理文件。此实现的另一个次要限制是，我们当前要等到缓冲区被填满后才能写入文件，如果没有新的推文出现，理论上可能会花费很长时间。

RawTweetsListener的代码如下所示：

from six import iteritems
import json
import csv
from tweepy.streaming import StreamListener
class RawTweetsListener(StreamListener):
    def __init__(self):
        self.buffered_data = []
        self.counter = 0

    def flush_buffer_if_needed(self):
        "Check the buffer capacity and write to a new file if needed"
        length = len(self.buffered_data)
        if length > 0 and length % 10 == 0:
            with open(os.path.join( output_dir,
                "tweets{}.csv".format(self.counter)), "w") as fs:
                self.counter += 1
                csv_writer = csv.DictWriter( fs,
                    fieldnames = fieldnames)
                for data in self.buffered_data:
 csv_writer.writerow(data)
            self.buffered_data = []

    def on_data(self, data):
        def transform(key, value):
            return transforms[key](value) if key in transforms else value

        self.buffered_data.append(
            {key:transform(key,value) \
                 for key,value in iteritems(json.loads(data)) \
                 if key in fieldnames}
        )
        self.flush_buffer_if_needed()
        return True

    def on_error(self, status):
        print("An error occured while receiving streaming data: {}".format(status))
        return False

Files

07.md

Latest commit

History

07.md

File metadata and controls

七、大数据和 Twitter 情感分析

Apache Spark 入门

注意

Apache Spark 架构

配置笔记本来配合 Spark

注意

注意

注意

Twitter 情感分析应用

第 1 部分——使用 Spark 结构化流获取数据

注意

数据管道的架构图

通过 Twitter 执行认证

提示

创建 Twitter 流

注意

注意

注意

注意

注意

创建 Spark 流数据框架

注意

注意

注意

创建并运行结构化查询

注意

注意

注意

监视活动的流查询

注意

注意

从 Parquet 文件创建批量DataFrame

注意

注意

注意

第 2 部分——使用情感和最相关的提取实体丰富数据

IBM Watson Natural 语言理解服务入门

注意

注意

注意

注意

注意

注意

注意

注意

注意

注意

第 3 部分——创建实时仪表板 PixieApp

注意

将分析重构为自己的方法

注意

注意

注意

注意

创建 PixieApp

注意

注意

注意

注意

注意

注意

注意

注意

注意

第 4 部分——使用 Apache Kafka 和 IBM Streams Designer 添加可伸缩性

注意

注意

将原始推文流式传输到 Kafka

注意

注意

注意

注意

注意

注意

从 Parquet 文件创建批量`DataFrame`

使用 Kafka 输入源创建 Spark Streaming `DataFrame`