Construindo pipelines de dados em R

É fácil argumentar que pipelines de ETL bem feitos são um órgão vital da ciência de dados. Sem dados limpos e organizados, fica difícil gerar insights de qualidade que orientem decisões de negócio.

Por isso, neste tutorial, vamos ver como construir um pipeline simples de ETL para transmitir Tweets em tempo real diretamente para um banco de dados SQLite usando R. Essa é uma tarefa bem comum em análises de redes sociais, por exemplo.

O foco será mostrar o processo de coleta e armazenamento de dados e como operar a API do Twitter usando o pacote rtweet em R.

Para começar, vamos garantir que você tenha as ferramentas certas. Primeiro, é preciso configurar seu acesso à API do Twitter. Em linhas gerais, você deve seguir estes passos:

• Criar uma conta no Twitter, caso ainda não tenha.
• Acessar este link e solicitar uma conta de desenvolvedor (observe que hoje o processo exige uma aplicação que precisa ser aprovada pelo Twitter).
• Criar um novo app nesta página.
• Preencher os detalhes do seu app e gerar seu access token.
• Guardar seu consumer key, consumer secret, access token e access token secret, que serão usados para conectar à API.

Com o acesso à API do Twitter pronto, você precisa instalar o SQLite caso ainda não tenha. Para ver o processo completo de instalação, siga o tutorial Guia para iniciantes em SQLite aqui na DataCamp. Optamos por usar SQLite neste tutorial pela simplicidade de operação.

Passo 1: criando um banco e uma tabela para armazenar os dados do Twitter

Com o acesso à API do Twitter e o SQLite instalados, podemos finalmente começar a construir um pipeline para armazenar Tweets à medida que os transmitimos ao longo do tempo. Para começar, vamos criar um novo banco SQLite usando R, como mostrado abaixo:

# 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")

Em seguida, podemos criar uma tabela dentro do banco para armazenar os tweets. No nosso caso, vamos guardar as seguintes variáveis:

• Tweet_ID como INTEGER primary key
• User como TEXT
• Tweet_Content como TEXT
• Date_Created como INTEGER

Você pode se perguntar: por que salvar as datas como inteiros? Porque o SQLite não tem um tipo de dado reservado para datas e horários. Assim, as datas serão armazenadas como o número de segundos desde 1970-01-01.

Agora, vamos criar a tabela:

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

Depois de criar a tabela, você pode abrir o sqlite3.exe e verificar se ela foi realmente criada. Veja a captura de tela abaixo:

Passo 2: transmita tweets sobre seus temas favoritos!

Acredite se quiser, os requisitos e a infraestrutura para ter um pipeline simples e funcional de streaming do Twitter já estão prontos. Agora precisamos transmitir os Tweets usando a API. Vale destacar que, para este tutorial, vou usar a API padrão gratuita. Existem versões pagas premium que podem atender melhor às suas necessidades de streaming de Tweets se você estiver conduzindo pesquisas, por exemplo.

Sem mais demora, vamos configurar nosso listener do Twitter. Primeiro, você precisa importar o pacote rtweet e informar os tokens e segredos de acesso do seu app, como descrevemos no início:

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')

Com o token pronto, o próximo passo é decidir quais tweets você quer transmitir (ou seja, ouvir). A função stream_tweets do pacote rtweet oferece várias opções para consultar a API do Twitter. Por exemplo, você pode transmitir tweets que contenham uma ou mais hashtags ou palavras-chave (até 400), uma pequena amostra aleatória de todos os tweets públicos, acompanhar os tweets de um grupo de IDs de usuário ou nomes de usuário (até 5000) ou coletar tweets por localização geográfica.

Neste tutorial, optei por transmitir tweets que contenham hashtags relacionadas à ciência de dados (veja a lista abaixo). Você vai notar que o formato em que defini as hashtags a transmitir é um pouco diferente, mas é o formato exigido pela função stream_tweets quando você quer ouvir hashtags ou palavras-chave. Esse formato muda se a escuta for por um conjunto de usuários ou por coordenadas. Para mais detalhes, consulte a documentação.

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

Com as keywords definidas, é hora de criar o loop de streaming. Há várias formas de fazer isso, mas este formato tem funcionado bem para mim:

# 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
}

Em essência, esse loop transmite o máximo possível de tweets que mencionem qualquer uma das hashtags da string de chaves em intervalos de 2 horas, por um total de 12 horas. A cada 2 horas, o listener cria um arquivo .json no diretório de trabalho atual com o nome definido na variável filename.

Depois, ele passa esse filename para a função transform_and_clean_tweets, que remove retweets, se desejado, seleciona as colunas que queremos manter dentre todas as fornecidas pela API do Twitter e normaliza o texto dos Tweets.

Em seguida, ele acrescenta o dataframe resultante à tabela Tweet_Data que criamos anteriormente no nosso banco SQLite. Por fim, soma 2 ao contador de horas (já que cada stream dura 2 horas) e remove o arquivo .json criado. Fazemos isso porque todos os dados de interesse já estão no banco, e manter os arquivos .json pode virar um problema de armazenamento.

Agora vamos olhar com mais atenção a função transform_and_clean_tweets em detalhes:

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)
}

Como vimos, essa função serve para filtrar retweets (se desejado), manter os campos necessários e normalizar o texto do tweet. Na prática, ela representa o "T" do acrônimo ETL (transformação). Um componente-chave aqui é o processo de limpeza do texto.

Normalmente, dados de texto exigem algumas etapas de pré-processamento antes de qualquer análise. No caso de tweets, isso pode incluir remoção de URLs, stopwords e menções, conversão para minúsculas, stemming, etc. Porém, nem sempre todas essas etapas são necessárias. Mesmo assim, por ora, mostro abaixo a função normalize_text que usei para pré-processar esses tweets:

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)
}

Dependendo do seu caso de uso, essas etapas podem ser suficientes. Como mencionei, você pode optar por adicionar outras etapas, como stemming ou lematização, ou manter apenas letras em vez de letras e números. Vale experimentar diferentes combinações. Esse também é um ótimo lugar para treinar regex.

Depois de tudo isso, o resultado é um banco SQLite populado com todos os tweets transmitidos. Você pode validar se tudo correu bem executando algumas consultas simples, por exemplo:

data_test <- dbGetQuery(conn, "SELECT * FROM Tweet_Data LIMIT 20")
unique_rows <- dbGetQuery(conn, "SELECT COUNT() AS Total FROM Tweet_Data")
kable(data_test)

print(as.numeric(unique_rows))

## [1] 1863

Passo 3: análise

Confirmado que o processo de ETL está funcionando como esperado, o passo final é extrair alguns insights e analisar os dados coletados. Com os tweets que reunimos, por exemplo, podemos fazer algumas coisas: uma nuvem de palavras com os termos mencionados no conteúdo dos tweets e uma linha do tempo para visualizar em que momento, dentro das 12 horas de streaming, obtivemos mais tweets. Claro, essa lista está longe de cobrir todo o leque de pesquisas possíveis com tweets, que vai de análise de sentimento a estudos psicográficos e além.

Dito isso, vamos direto à construção de uma boa wordcloud:

# 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"))

Pois é, que surpresa maravilhosa! machinelearning e datascience são as palavras mais mencionadas em todos os nossos tweets — justamente duas das hashtags que estávamos transmitindo. Ou seja, era esperado. As outras palavras, porém, são mais interessantes de observar. Por exemplo, bigdata e artificialintelligence não estavam nas nossas chaves, mas apareceram com alta frequência, indicando que são temas muito citados junto com as duas anteriores. Também surgiram termos como python e tensorflow, que dão mais contexto sobre o conteúdo dos tweets para além das hashtags.

Vamos agora para outra análise simples. No nosso stream de 12 horas, em que horário coletamos mais tweets? Para isso, vamos buscar as datas inteiras, convertê-las para um formato adequado e, então, plotar a quantidade de tweets ao longo do tempo:

# 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")

Perfeito! Agora dá para ver que o maior volume de tweets únicos veio do dia 2 de setembro, entre 20:00 e 20:59 UTC (horário militar: 20).

Conclusão

Parabéns! Agora você sabe como construir um pipeline simples de ETL em R. As duas análises que fizemos são bem básicas usando dados do Twitter. Mas, como comentado, há muito mais a fazer, desde que você construa um pipeline robusto para trazer os dados. Esse foi o principal foco deste tutorial.

Dito isso, este tutorial mostrou apenas um estudo de caso em pequena escala para percorrer o processo de construção de pipelines de ETL para dados do Twitter. Criar pipelines de ETL robustos e escaláveis para toda uma empresa é um esforço complexo, que exige muitos recursos computacionais e conhecimento, especialmente quando entra big data na jogada.

Eu te incentivo a pesquisar mais e tentar construir seus próprios pipelines em pequena escala — pode ser em Python, por exemplo. Quem sabe até se aventurar em projetos de big data. A DataCamp já tem cursos como o Big Data Fundamentals via PySpark, que aborda big data com ferramentas como PySpark para aprofundar seus conhecimentos na área.

Se você quer aprender sobre engenharia de dados, faça o curso Introduction to Data Engineering da DataCamp. E se já estiver pronto para provar suas novas habilidades para recrutadores, confira nossa Data Engineer Certification. 

Referências

1. Foley, D. (2019, 11 de maio). Streaming Twitter Data into a MySQL Database. Recuperado de https://towardsdatascience.com/streaming-twitter-data-into-a-mysql-database-d62a02b050d6