跳至主要内容

使用 R 构建数据管道

学习如何从零开始使用 R 和 SQLite 构建 ETL 数据管道，以实时收集推文并将其存储起来用于后续分析。
· 13分钟 阅读

使用 AI 探索

可以说，规范的 ETL 管道是数据科学的“生命器官”之一。没有干净、有序的数据，就很难产出能提升业务决策的高质量洞察。

因此，在本教程中，我们将探索如何使用 R 将实时推文直接流式写入 SQLite 数据库，以构建一个简单的 ETL 管道。例如，这在社交网络分析中就是一种相当常见的任务。

本文将重点介绍数据收集与存储的思路，以及如何在 R 中使用 rtweet 包操作 Twitter API。

首先，准备好所有必需工具。您需要做的第一件事是设置对 Twitter API 的访问。总体而言，您需要按照以下步骤进行：

  • 如果您还没有 Twitter 帐号，请先创建一个。
  • 通过此 链接 申请开发者账号（请注意，现在需要提交申请并等待 Twitter 审批）。
  • 在以下 网页 创建一个新应用。
  • 填写应用的所有信息并创建访问令牌。
  • 收集您的 consumer key、consumer secret、access token 和 access token secret，这些是连接 API 所必需的。

完成 Twitter API 的相关设置后，如果您还没有 SQLite，则需要获取并安装。有关在计算机上安装 SQLite 的完整流程，请参阅 DataCamp 的 SQLite 初学者指南。本教程选择 SQLite，主要是因为它操作简单。

步骤 1：创建用于存储 Twitter 数据的数据库与数据表

在准备好 Twitter API 访问权限并安装 SQLite 之后，我们终于可以开始构建一个随时间流式存储推文的管道。首先，我们将按如下所示使用 R 创建一个新的 SQLite 数据库：

# Import necessary libraries and functions
library(RSQLite)
library(rtweet)
library(tm)
library(dplyr)
library(knitr)
library(wordcloud)
library(lubridate)
library(ggplot2)
source("transform_and_clean_tweets.R")

# Create our SQLite database
conn <- dbConnect(RSQLite::SQLite(), "Tweet_DB.db")

接下来，我们可以在数据库中创建一个表来保存推文。在我们的示例中，将存储以下变量：

  • Tweet_ID，作为 INTEGER 主键
  • User，为 TEXT
  • Tweet_Content，为 TEXT
  • Date_Created，为 INTEGER

您可能会疑惑：为什么将日期设置为整数？这是因为 SQLite 没有专门的日期和时间数据类型。因此，日期将以自 1970-01-01 起的秒数进行存储。

现在来编写创建数据表的语句：

dbExecute(conn, "CREATE TABLE Tweet_Data(
                  Tweet_ID INTEGER PRIMARY KEY,
                  User TEXT,
                  Tweet_Content TEXT,
                  Date_Created INTEGER)")

创建数据表后，您可以打开 sqlite3.exe 检查是否已经创建成功。如下图所示：

sqlite3.exe 截图


步骤 2：流式采集您感兴趣主题的推文！

信不信由您，至此，搭建一个简单可用的 Twitter 流式管道所需的条件与基础设施就都齐备了。我们现在需要做的是通过 API 来流式采集推文。值得一提的是，本教程使用的是标准的免费 API。如果您进行研究等工作，付费的高级版本可能更能满足您的流式需求。

事不宜迟，我们开始设置 Twitter 监听器。首先，您需要导入 rtweet 包，并按开头所述输入应用的访问令牌和密钥：

token <- create_token(app = 'Your_App_Name',
                      consumer_key = 'Your_Consumer_Key',
                      consumer_secret = 'Your_Consumer_Secret',
                      access_token = 'Your_Access_Token',
                      access_secret = 'Your_Access_Secret')

配置好令牌后，下一步就是决定要流式采集哪些推文（即您要“监听”什么）。rtweet 包中的 stream_tweets 函数为我们提供了多种查询 Twitter API 的方式。例如，您可以流式采集包含给定的一组标签或关键词（最多 400 个）之一的推文、所有公开推文中的小部分随机子集、跟踪一组用户 ID 或用户名（最多 5000 个）的推文，或者按地理位置采集推文。

在本教程中，我决定流式采集包含与数据科学相关标签的推文（见下方列表）。您可能注意到，我定义要监听的标签时使用的格式有点奇怪，但当您要监听标签或关键词时，这正是 stream_tweets 函数所要求的格式。如果您打算监听指定用户或基于坐标，格式会有所不同。更多详情请参阅其文档。

keys <- "#nlp,#machinelearning,#datascience,#chatbots,#naturallanguageprocessing,#deeplearning"

定义好关键词后，就可以编写推文流式采集的循环了。实现方式有很多种，下面这种写法在我过去的实践中表现不错：

# Initialize the streaming hour tally
hour_counter <- 0

# Initialize a while loop that stops when the number of hours you want to stream tweets for is exceeded
while(hour_counter <= 12){
  # Set the stream time to 2 hours each iteration (7200 seconds)
  streamtime <- 7200
  # Create the file name where the 2 hour stream will be stored. Note that the Twitter API outputs a .json file.
  filename <- paste0("nlp_stream_",format(Sys.time(),'%d_%m_%Y__%H_%M_%S'),".json")
  # Stream Tweets containing the desired keys for the specified amount of time
  stream_tweets(q = keys, timeout = streamtime, file_name = filename)
  # Clean the streamed tweets and select the desired fields
  clean_stream <- transform_and_clean_tweets(filename, remove_rts = TRUE)
  # Append the streamed tweets to the Tweet_Data table in the SQLite database
  dbWriteTable(conn, "Tweet_Data", clean_stream, append = T)
  # Delete the .json file from this 2-hour stream
  file.remove(filename)
  # Add the hours to the tally
  hour_counter <- hour_counter + 2
}

本质上，该循环会在总计 12 小时的时间里，每隔 2 小时尽可能多地流式获取包含关键字字符串中任一标签的推文。每 2 小时，Twitter 监听器都会在您当前工作目录下创建一个以变量 filename 指定名称命名的 .json 文件。

然后，它将该文件名传递给 transform_and_clean_tweets 函数，以便在需要时移除转推、从 Twitter API 返回的大量字段中选择我们需要保留的列，并对推文文本进行规范化。

接着，它会将得到的数据框追加到我们先前在 SQLite 数据库中创建的 Tweet_Data 表中。最后，由于每次流式为 2 小时，它会将计数累加 2，并删除创建的 .json 文件。之所以删除，是因为我们所需的数据已经进入数据库，保留这些 .json 文件会占用大量存储空间。

现在我们更详细地看看 transform_and_clean_tweets 函数：

transform_and_clean_tweets <- function(filename, remove_rts = TRUE){

  # Import the normalize_text function
  source("normalize_text.R")

  # Parse the .json file given by the Twitter API into an R data frame
  df <- parse_stream(filename)
  # If remove_rst = TRUE, filter out all the retweets from the stream
  if(remove_rts == TRUE){
    df <- filter(df,df$is_retweet == FALSE)
  }
  # Keep only the tweets that are in English
  df <- filter(df, df$lang == "en")
  # Select the features that you want to keep from the Twitter stream and rename them
  # so the names match those of the columns in the Tweet_Data table in our database
  small_df <- df[,c("screen_name","text","created_at")]
  names(small_df) <- c("User","Tweet_Content","Date_Created")
  # Finally normalize the tweet text
  small_df$Tweet_Content <- sapply(small_df$Tweet_Content, normalize_text)
  # Return the processed data frame
  return(small_df)
}

如上所述，该函数用于在需要时过滤转推、保留所需特征并规范化推文文本。从本质上讲，这可以视作 ETL 缩写中的 “T”（转换）部分。该函数的一个关键环节是清洗推文文本。

通常，文本数据在可用于任何形式的分析之前，都需要进行一些预处理步骤。对于推文，这些步骤可能包括去除 URL、停用词与提及，将文本转换为小写、词干提取等。不过，并非所有步骤在任何时候都必需。下面我先给出用于预处理这些推文的 normalize_text 函数：

normalize_text <- function(text){
  # Keep only ASCII characters
  text = iconv(text, "latin1", "ASCII", sub="")
  # Convert to lower case characters
  text = tolower(text)
  # Remove any HTML tags
  text = gsub("<.*?>", " ", text)
  # Remove URLs
  text = gsub("\\s?(f|ht)(tp)(s?)(://)([^\\.]*)[\\.|/](\\S*)", "", text)
  # Keep letters and numbers only
  text = gsub("[^[:alnum:]]", " ", text)
  # Remove stop words
  text = removeWords(text,c("rt","gt",stopwords("en")))
  # Remove any extra white space
  text = stripWhitespace(text)                                 
  text = gsub("^\\s+|\\s+$", "", text)                         

  return(text)
}

根据您的使用场景，上述步骤可能已经足够。正如之前所提，您也可以加入更多或不同的步骤，例如词干提取或词形还原，或者只保留字母而非字母与数字。我鼓励您多做尝试，测试不同组合。这也是练习正则表达式技能的好机会。

完成以上步骤后，我们将得到一个已填充所有流式推文的 SQLite 数据库。您可以通过运行几个简单查询来验证是否一切正常，例如：

data_test <- dbGetQuery(conn, "SELECT * FROM Tweet_Data LIMIT 20")
unique_rows <- dbGetQuery(conn, "SELECT COUNT() AS Total FROM Tweet_Data")
kable(data_test)
填充了所有流式推文的 SQLite 数据库
print(as.numeric(unique_rows))

## [1] 1863


步骤 3：分析

当可以确定 ETL 过程按预期工作后，最后一步就是从收集到的数据中提炼一些洞察并进行分析。以我们收集到的推文为例，可以尝试做几件事：基于推文内容的词云，以及一条时间线，用来可视化在 12 小时流式期间我们在哪些时间段获取了最多推文。当然，这份清单并不全面，围绕推文可开展的研究范围很广，从情感分析到心理画像研究等不一而足。

话不多说，我们先来绘制一个漂亮的词云：

# Gather all tweets from the database
all_tweets <- dbGetQuery(conn, "SELECT Tweet_ID, Tweet_Content FROM Tweet_Data")

# Create a term-document matrix and sort the words by frequency
dtm <- TermDocumentMatrix(VCorpus(VectorSource(all_tweets$Tweet_Content)))
dtm_mat <- as.matrix(dtm)
sorted <- sort(rowSums(dtm_mat), decreasing = TRUE)
freq_df <- data.frame(words = names(sorted), freq = sorted)

# Plot the wordcloud
set.seed(42)
wordcloud(words = freq_df$words, freq = freq_df$freq, min.freq = 10,
          max.words=50, random.order=FALSE, rot.per=0.15,
          colors=brewer.pal(8, "RdYlGn"))
推文词云


哦，这并不出乎意料！machinelearning 和 datascience 是我们所有推文中提及最多的词——它们正是我们所监听的两个标签。因此，这个结果是意料之中的。其他词就更有意思了。例如，bigdata 和 artificialintelligence 并不在我们的监听标签中，但它们出现频率很高，可以认为它们经常与前两者一起被讨论。我们还捕捉到其他有价值的词，比如 python 或 tensorflow，这为我们理解所收集推文的内容提供了更多超越标签本身的上下文。

接下来进行另一项简单分析：在这 12 小时的流式过程中，我们在哪个时间点收集到的推文最多？为此，我们将提取整数时间戳，将其转换为合适的格式，然后绘制推文数量随时间变化的图表：

# Select the dates in which the tweets were created and convert them into UTC date-time format
all_tweets <- dbGetQuery(conn, "SELECT Tweet_ID, Date_Created FROM Tweet_Data")
all_tweets$Date_Created <- as.POSIXCT(all_tweets$Date_Created, origin = "1970-01-01", tz = "UTC")

# Group by the day and hour and count the number of tweets that occurred in each bucket
all_tweets_2 <- all_tweets %>%
    mutate(day = day(Date_Created),
           month = month(Date_Created, label = TRUE),
           hour = hour(Date_Created)) %>%
    mutate(day_hour = paste(month,"-",day,"-",hour, sep = "")) %>%
    group_by(day_hour) %>%
    tally()

# Simple line ggplot
ggplot(all_tweets_2, aes(x = day_hour, y = n)) +
  geom_line(aes(group = 1)) +
  theme_minimal() +
  ggtitle("Tweet Freqeuncy During the 12-h Streming Period")+
  ylab("Tweet Count")+
  xlab("Month-Day-Hour")
推文数量随时间变化的图表


很好！现在我们可以看到，获取到的唯一推文数量最多的时间段是 9 月 2 日 UTC 时间晚上 8:00 至 8:59（以 24 小时制显示为 20）。

总结

恭喜您！现在您已经学会如何在 R 中构建一个简单的 ETL 管道。我们进行的两项分析都属于使用 Twitter 数据的入门范例。不过，如前所述，只要您搭建起稳健的入数管道，可开展的工作还有很多。

需要说明的是，本文只是通过一个小规模的案例带您走一遍为 Twitter 数据构建 ETL 管道的流程。为整个企业构建稳健且可扩展的 ETL 管道是一项复杂工作，特别是在涉及大数据时，需要大量计算资源与专业知识。

我鼓励您继续研究，并尝试构建属于您自己的小规模管道，也可以尝试用 Python 搭建。甚至可以挑战一些大数据项目。例如，DataCamp 已推出类似课程——通过 PySpark 学习大数据基础，使用 PySpark 等工具深入讲解大数据，能够进一步拓展您的相关知识。

如果您想学习数据工程，请学习 DataCamp 的 数据工程入门 课程。若您已准备好向雇主证明新技能，欢迎查看我们的 数据工程师认证。


参考资料

  1. Foley, D. (2019 年 5 月 11 日)。将 Twitter 数据流式写入 MySQL 数据库。检索自 https://towardsdatascience.com/streaming-twitter-data-into-a-mysql-database-d62a02b050d6

主题

R数据科学数据工程机器学习SQL