Corso
Fondamenti di Big Data con PySpark
4 ore
66.8K
Si potrebbe sostenere che le pipeline ETL ben fatte siano un organo vitale della data science. Senza dati puliti e organizzati, diventa difficile produrre insight di qualità che migliorino le decisioni di business.
In questo tutorial, quindi, vedremo cosa comporta costruire una semplice pipeline ETL per trasmettere in streaming tweet in tempo reale direttamente in un database SQLite usando R. Questo è un compito piuttosto comune, ad esempio, nelle analisi dei social network.
Ci concentreremo sul processo di raccolta e archiviazione dei dati e su come usare l’API di Twitter con il pacchetto rtweet in R.
Per cominciare, procuriamoci tutti gli strumenti giusti. La prima cosa di cui avrai bisogno è configurare l’accesso all’API di Twitter. In generale, dovrai seguire questi passaggi:
- Crea un account Twitter, se non ne hai già uno.
- Segui questo link e richiedi un account sviluppatore (nota che ora questo processo richiede una domanda che Twitter deve approvare).
- Crea una nuova app sulla seguente pagina web.
- Compila tutti i dettagli della tua app e crea il tuo access token.
- Recupera consumer key, consumer secret, access token e access token secret, necessari per connetterti all’API.
Una volta impostato l’accesso all’API di Twitter, procurati SQLite se non lo hai già. Per l’intera procedura su come installare SQLite sul tuo computer, segui il tutorial Guida per principianti a SQLite qui su DataCamp. La scelta di usare SQLite in questo tutorial è dovuta alla sua semplicità d’uso.
Passaggio 1: creare un database e una tabella per archiviare i dati di Twitter
Con l’accesso all’API di Twitter e SQLite installato, possiamo finalmente iniziare a costruire una pipeline per salvare i tweet man mano che li trasmettiamo nel tempo. Per iniziare, creeremo il nuovo database SQLite usando R come indicato di seguito:
# 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")
Successivamente, possiamo creare una tabella all’interno del database per contenere i tweet. Nel nostro caso specifico memorizzeremo le seguenti variabili:
- Tweet_ID come INTEGER chiave primaria
- User come TEXT
- Tweet_Content come TEXT
- Date_Created come INTEGER
Potresti chiederti: perché imposto le date come interi? Perché SQLite non ha un tipo di dato riservato per date e orari. Pertanto, le date verranno salvate come numero di secondi dal 1970-01-01.
Procediamo ora a scrivere la tabella:
dbExecute(conn, "CREATE TABLE Tweet_Data(
Tweet_ID INTEGER PRIMARY KEY,
User TEXT,
Tweet_Content TEXT,
Date_Created INTEGER)")
Dopo aver creato la tabella, puoi aprire sqlite3.exe e verificare che sia effettivamente stata creata. Ecco uno screenshot di esempio:
Passaggio 2: trasmetti in streaming tweet sui tuoi argomenti preferiti!
Per quanto possa sembrare incredibile, hai finito con i requisiti e l’infrastruttura necessari per avere una pipeline di streaming Twitter semplice e funzionante. Ora dobbiamo trasmettere in streaming i tweet usando l’API. Vale la pena notare che, per gli scopi di questo tutorial, userò l’API standard gratuita. Esistono versioni premium a pagamento che potrebbero adattarsi meglio alle tue esigenze di streaming se, ad esempio, conduci attività di ricerca.
Senza ulteriori indugi, iniziamo a configurare il nostro listener di Twitter. Per prima cosa dovrai importare il pacchetto rtweet e inserire i token di accesso e i secret della tua applicazione come descritto all’inizio:
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 il tuo token a posto, la prossima cosa è decidere quali tweet vuoi trasmettere in streaming (cioè ascoltare). La funzione stream_tweets del pacchetto rtweet ci offre diverse opzioni per interrogare l’API di Twitter. Ad esempio, puoi trasmettere in streaming tweet che contengono uno o più tra un insieme di hashtag o parole chiave (fino a 400), un piccolo sottoinsieme casuale di tutti i tweet pubblici disponibili, tracciare i tweet di un gruppo di ID utente o screen name (fino a 5000) o raccogliere tweet per posizione geografica.
Per questo tutorial, ho deciso di trasmettere in streaming i tweet contenenti hashtag legati alla data science (vedi elenco sotto). Potresti notare che il formato in cui ho definito gli hashtag da ascoltare è un po’ insolito. Tuttavia, è il formato richiesto dalla funzione stream_tweets quando vuoi ascoltare hashtag o parole chiave. Questo formato varia se intendi ascoltare un dato insieme di utenti o usare coordinate geografiche. Per maggiori dettagli, consulta la documentazione.
keys <- "#nlp,#machinelearning,#datascience,#chatbots,#naturallanguageprocessing,#deeplearning"
Definite le parole chiave, è il momento di impostare il ciclo di streaming dei tweet. Ci sono diversi modi per farlo, ma questo formato in passato ha funzionato bene per me:
# 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
}
In sostanza, quel ciclo trasmette quanti più tweet possibile che menzionano uno qualsiasi degli hashtag nella stringa delle chiavi a intervalli di 2 ore per un tempo totale di 12 ore. Ogni 2 ore, il listener Twitter crea un file .json nella tua directory di lavoro corrente con il nome specificato nella variabile filename.
Quindi passa questo filename alla funzione transform_and_clean_tweets, che rimuove i retweet se desiderato, seleziona le colonne che vogliamo mantenere tra tutte quelle fornite dall’API di Twitter e normalizza il testo contenuto nei tweet.
Poi aggiunge il dataframe risultante alla tabella Tweet_Data che abbiamo creato in precedenza all’interno del nostro database SQLite. Infine, aggiunge 2 al conteggio delle ore (dato che ogni stream dura 2 ore) e rimuove il file .json creato. Questo perché tutti i dati di interesse sono ora nel database e mantenere i file .json può diventare oneroso in termini di spazio.
Vediamo ora più da vicino la funzione transform_and_clean_tweets nel dettaglio:
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)
}
Come detto, questa funzione serve a filtrare i retweet se necessario, mantenere le caratteristiche desiderate e normalizzare il testo dei tweet. In sostanza, può essere considerata la parte "T" dell’acronimo ETL (la trasformazione). Un elemento chiave di questa funzione è il processo di pulizia del testo dei tweet.
Di solito, i dati testuali richiedono alcuni passaggi di pre-processing prima di essere pronti per qualsiasi tipo di analisi. Nel caso dei tweet, questi passaggi possono includere la rimozione di URL, stop word e menzioni, la conversione del testo in minuscolo, lo stemming delle parole, ecc. Tuttavia, non tutti questi passaggi sono sempre necessari. Detto ciò, riporto di seguito la funzione normalize_text che ho usato per pre-processare questi tweet:
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)
}
A seconda del tuo caso d’uso, questi passaggi potrebbero essere sufficienti. Come accennato, potresti scegliere di aggiungere passaggi diversi, come stemming o lemmatizzazione, o mantenere solo le lettere invece di lettere e numeri. Ti incoraggio a sperimentare combinazioni diverse. È anche un ottimo modo per fare pratica con le regex.
Dopo tutti questi passaggi, il risultato è un database SQLite popolato con tutti i tweet trasmessi in streaming. Puoi verificare che tutto abbia funzionato correttamente eseguendo, ad esempio, un paio di semplici query:
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
Passaggio 3: analizza
Quando sei certo che il processo ETL funzioni come previsto, l’ultimo passaggio è ricavare alcuni insight e analizzare i dati raccolti. Ad esempio, per i tweet che abbiamo raccolto, possiamo provare un paio di cose diverse: una word cloud dei termini menzionati nel contenuto dei tweet e una timeline per visualizzare quando, nel nostro periodo di streaming di 12 ore, abbiamo ottenuto la maggiore quantità di tweet. Questo elenco non è ovviamente esaustivo rispetto all’ampia gamma di ricerche possibili con i tweet, che spaziano dall’analisi del sentiment a studi psicografici e oltre.
Detto ciò, passiamo subito a costruire una bella word cloud:
# 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"))
Beh, che enorme sorpresa! machinelearning e datascience sono le parole più menzionate in tutti i nostri tweet: guarda caso sono due degli hashtag che stavamo monitorando. Dunque, questo risultato era atteso. Le altre parole però sono un po’ più interessanti da osservare. Ad esempio, bigdata e artificialintelligence non erano hashtag nelle nostre chiavi, eppure sono comparse con alta frequenza, quindi si potrebbe sostenere che vengano spesso discusse insieme alle altre due. Ci sono anche altre parole interessanti che abbiamo raccolto, come python o tensorflow, che ci danno un po’ più di contesto sul contenuto dei tweet raccolti, oltre ai soli hashtag.
Passiamo ora a un’altra semplice analisi. Nel nostro stream di 12 ore, a che ora abbiamo raccolto il maggior numero di tweet? Per farlo, recupereremo le date intere, le trasformeremo in un formato corretto e poi tracceremo la quantità di tweet nel 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")
Ottimo! Ora possiamo vedere che la quantità maggiore di tweet unici è stata ottenuta il 2 settembre tra le 20:00 UTC e le 20:59 UTC (compare in orario militare come 20).
Conclusione
Congratulazioni! Ora sai come costruire una semplice pipeline ETL in R. Le due analisi che abbiamo effettuato rappresentano analisi di base condotte con dati di Twitter. Tuttavia, come già detto, c’è moltissimo da fare, purché tu costruisca una pipeline solida per acquisire i dati. Questo è stato l’aspetto principale su cui si è concentrato il tutorial.
Detto questo, il tutorial ha mostrato solo un piccolo caso di studio per ripercorrere il processo di costruzione di pipeline ETL per dati di Twitter. Creare pipeline ETL robuste e scalabili per un’intera azienda è un’impresa complessa che richiede ampie risorse computazionali e competenze, soprattutto quando è coinvolta la big data.
Ti incoraggio a fare ulteriori ricerche e provare a costruire piccole pipeline, magari anche in Python. Puoi persino cimentarti in progetti di big data. Ad esempio, DataCamp ha già corsi come Fondamenti di Big Data con PySpark, in cui si affronta il big data con strumenti come PySpark per approfondire le tue conoscenze in questo campo.
Se vuoi imparare l’ingegneria dei dati, segui il corso di DataCamp Introduzione al Data Engineering. E se sei pronto a dimostrare le tue nuove competenze ai datori di lavoro, scopri la nostra Certificazione Data Engineer.
Riferimenti
- Foley, D. (11 maggio 2019). Streaming Twitter Data into a MySQL Database. Recuperato da https://towardsdatascience.com/streaming-twitter-data-into-a-mysql-database-d62a02b050d6
