Creación de canalizaciones de datos con R

Se podría decir que las canalizaciones ETL bien diseñadas son un órgano vital de la ciencia de datos. Sin datos limpios y organizados, es difícil generar insights de calidad que impulsen decisiones de negocio.

Por eso, en este tutorial veremos cómo crear una canalización ETL sencilla para transmitir Tweets en tiempo real directamente a una base de datos SQLite usando R. Es una tarea bastante habitual en análisis de redes sociales, por ejemplo.

Nos centraremos en el proceso de recogida y almacenamiento de datos y en cómo operar la API de Twitter con el paquete rtweet en R.

Para empezar, tengamos a mano todas las herramientas. Lo primero que necesitas es configurar tu acceso a la API de Twitter. En líneas generales, debes seguir estos pasos:

• Crea una cuenta de Twitter, si aún no la tienes.
• Sigue este enlace y solicita una cuenta de desarrollador (ten en cuenta que ahora Twitter debe aprobar tu solicitud).
• Crea una app nueva en la siguiente página.
• Rellena todos los datos de tu app y crea tu token de acceso.
• Guarda tu consumer key, consumer secret, access token y access token secret, que necesitas para conectar con la API.

Cuando tengas listo el acceso a la API de Twitter, instala SQLite si no lo tienes. Para ver el proceso completo de instalación en tu equipo, sigue el tutorial Guía para principiantes de SQLite aquí en DataCamp. Elegimos SQLite para este tutorial por su sencillez.

Paso 1: crear una base de datos y una tabla para almacenar los datos de Twitter

Con el acceso a la API de Twitter y SQLite instalado, por fin podemos empezar a construir una canalización para almacenar Tweets a medida que los recibimos con el tiempo. Para empezar, vamos a crear la nueva base de datos SQLite desde R como se indica a continuación:

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

A continuación, crearemos una tabla dentro de la base de datos para guardar los tweets. En nuestro caso, almacenaremos las siguientes variables:

• Tweet_ID como clave primaria INTEGER
• User como TEXT
• Tweet_Content como TEXT
• Date_Created como INTEGER

Quizá te preguntes por qué guardo las fechas como enteros. Esto se debe a que SQLite no tiene un tipo de datos reservado para fechas y horas. Por tanto, las fechas se almacenarán como el número de segundos desde 1970-01-01.

Ahora vamos a crear la tabla:

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

Una vez creada la tabla, puedes ir a sqlite3.exe y comprobar que efectivamente se ha creado. Debajo puedes ver una captura:

Paso 2: transmite tweets sobre tus temas favoritos

Créelo o no, ya tienes listos los requisitos y la infraestructura para una canalización de streaming de Twitter sencilla y funcional. Ahora toca transmitir tweets usando la API. Ten en cuenta que para este tutorial utilizaré la API estándar gratuita. Existen versiones premium de pago que quizá se adapten mejor a tus necesidades de streaming si, por ejemplo, realizas investigación.

Sin más, vamos a configurar nuestro listener de Twitter. Primero, importa el paquete rtweet e introduce los tokens y secretos de acceso de tu aplicación como describimos al principio:

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

Con el token listo, lo siguiente es decidir qué tweets quieres transmitir (es decir, escuchar). La función stream_tweets del paquete rtweet ofrece varias opciones para consultar la API de Twitter. Por ejemplo, puedes transmitir tweets que contengan uno o varios hashtags o palabras clave (hasta 400), una pequeña muestra aleatoria de todos los tweets públicos, seguir los tweets de un grupo de IDs de usuario o nombres de pantalla (hasta 5000), o recopilar tweets por ubicación geográfica.

Para este tutorial, he decidido transmitir tweets que contengan hashtags relacionados con ciencia de datos (ver lista abajo). Verás que el formato en el que defino los hashtags a transmitir es algo peculiar. Sin embargo, es el formato que requiere la función stream_tweets cuando quieres escuchar hashtags o palabras clave. Este formato varía si pretendes escuchar a un conjunto de usuarios o por coordenadas. Para más detalles, consulta la documentación.

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

Con las palabras clave definidas, toca crear el bucle de streaming. Hay varias formas de hacerlo, pero este formato me ha funcionado bien en el pasado:

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

En esencia, este bucle transmite tantos tweets como sea posible que mencionen cualquiera de los hashtags en la cadena de claves en intervalos de 2 horas hasta un total de 12 horas. Cada 2 horas, el listener de Twitter crea un archivo .json en tu directorio de trabajo actual con el nombre especificado en la variable filename.

Después, pasa ese nombre de archivo a la función transform_and_clean_tweets, que elimina los retweets si así lo deseas, selecciona las columnas que queremos conservar de todas las que devuelve la API de Twitter y normaliza el texto contenido en los Tweets.

Luego, añade el dataframe resultante a la tabla Tweet_Data que creamos antes en nuestra base de datos SQLite. Por último, suma 2 al contador de horas (ya que cada transmisión dura 2 horas) y elimina el archivo .json creado. Hacemos esto porque todos los datos de interés ya están en la base de datos, y guardar los .json puede convertirse en un problema de almacenamiento.

Ahora veamos con más detalle la función 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)
}

Como ves, esta función sirve para filtrar retweets si se desea, conservar las variables relevantes y normalizar el texto del tweet. En esencia, puede considerarse la parte "T" del acrónimo ETL (la transformación). Un componente clave es el proceso de limpieza del texto.

Normalmente, los datos de texto requieren varios pasos de preprocesado antes de poder analizarlos. En el caso de los tweets, estos pasos pueden incluir eliminar URLs, stopwords y menciones, pasar a minúsculas, stemming, etc. No obstante, no siempre son necesarios todos. A continuación te muestro la función normalize_text que utilicé para preprocesar estos 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)
}

Según tu caso de uso, estos pasos pueden ser suficientes. Como comentaba, puedes añadir otros como stemming o lematización, o quedarte solo con letras en lugar de letras y números. Te animo a experimentar con distintas combinaciones. Es un buen lugar para practicar regex.

Tras todos estos pasos, el resultado es una base de datos SQLite poblada con todos los tweets transmitidos. Puedes comprobar que todo ha ido bien ejecutando un par de consultas sencillas, por ejemplo:

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

Paso 3: analiza

Cuando tengas la seguridad de que el proceso ETL funciona como debe, el paso final es extraer insights y analizar los datos recopilados. Por ejemplo, con los tweets que hemos reunido, podemos hacer un par de cosas: una nube de palabras con los términos mencionados en el contenido del tweet y una línea temporal para visualizar en qué momento de las 12 horas de streaming obtuvimos más tweets. Esta lista no es, obviamente, un repaso completo de todo lo que puede investigarse con tweets: desde análisis de sentimiento hasta estudios psicográficos y más allá.

Dicho esto, vayamos directos a crear una buena nube de palabras:

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

¡Vaya sorpresa! machinelearning y datascience son las palabras más mencionadas en todos nuestros tweets; justo dos de los hashtags que estábamos transmitiendo. Así que era de esperar. Las demás son más interesantes. Por ejemplo, bigdata y artificialintelligence no estaban en nuestras claves y, aun así, aparecen con alta frecuencia, por lo que cabe pensar que se habla de ellas a menudo junto a las otras dos. También captamos palabras como python o tensorflow, que nos dan algo más de contexto sobre el contenido de los tweets más allá de los hashtags.

Pasemos ahora a otro análisis sencillo. En nuestro stream de 12 horas, ¿a qué hora recopilamos más tweets? Para ello, tomaremos las fechas enteras, las convertiremos a un formato adecuado y representaremos la cantidad de tweets a lo largo del tiempo:

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

¡Genial! Ahora vemos que la mayor cantidad de tweets únicos que obtuvimos fue el 2 de septiembre entre las 20:00 y las 20:59 UTC (aparece en formato de 24 horas como 20).

Conclusión

¡Enhorabuena! Ya sabes cómo crear una canalización ETL sencilla en R. Los dos análisis que hemos hecho son muy básicos con datos de Twitter. Aun así, como comentábamos, hay muchísimo más que puedes hacer siempre que construyas una canalización robusta para traer los datos. Ese ha sido el foco principal de este tutorial.

Dicho esto, este tutorial solo muestra un caso a pequeña escala para recorrer el proceso de creación de canalizaciones ETL para datos de Twitter. Diseñar canalizaciones ETL robustas y escalables para toda una empresa es una tarea compleja que requiere amplios recursos de cómputo y conocimiento, especialmente cuando hablamos de big data.

Te animo a seguir investigando y a construir tus propias canalizaciones a pequeña escala, quizá en Python. Incluso puedes lanzarte con proyectos de big data. Por ejemplo, en DataCamp ya tienes cursos como Big Data Fundamentals via PySpark, donde se trabaja con big data usando herramientas como PySpark para profundizar en este campo.

Si te interesa aprender sobre ingeniería de datos, haz el curso Introduction to Data Engineering de DataCamp. Y si estás listo para demostrar tus nuevas habilidades a los empleadores, echa un vistazo a nuestra Data Engineer Certification.

Referencias

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