Datapijplijnen bouwen met R

Leer hoe je vanaf nul een ETL-datapijplijn opzet met R en SQLite om tweets in realtime te verzamelen en op te slaan voor toekomstig onderzoek.

Bijgewerkt 2 okt 2026 · 13 min lezen

Verken met AI

ChatGPT Claude Perplexity

Je zou kunnen stellen dat goede ETL-pijplijnen het kloppend hart zijn van data science. Zonder schone en geordende data wordt het lastig om kwalitatieve inzichten te produceren die bedrijfsbeslissingen verbeteren.

Daarom verkennen we in deze tutorial wat er komt kijken bij het bouwen van een eenvoudige ETL-pijplijn om realtime tweets rechtstreeks in een SQLite-database te streamen met R. Dit is bijvoorbeeld een vrij gangbare taak binnen socialenetwerkanalyse.

De focus ligt op het denkproces rond dataverzameling en -opslag, en op hoe je de Twitter API bedient met het rtweet-pakket in R.

Laten we beginnen met de juiste tools. Als eerste moet je toegang tot de Twitter API instellen. In grote lijnen volg je deze stappen:

Maak een Twitter-account aan als je dat nog niet hebt.
Volg deze link en vraag een ontwikkelaarsaccount aan (let op: dit vereist nu een aanvraag die door Twitter moet worden goedgekeurd).
Maak een nieuwe app op de volgende webpagina.
Vul alle details over je app in en maak je access token aan.
Verzamel je consumer key, consumer secret, access token en access token secret; deze heb je nodig om met de API te verbinden.

Als je klaar bent met de Twitter API, heb je SQLite nodig als je dat nog niet hebt. Voor het volledige proces om SQLite op je computer te installeren, volg je de Beginners Guide to SQLite-tutorial hier op DataCamp. De keuze voor SQLite in deze tutorial is gemaakt vanwege de eenvoud in gebruik.

Stap 1: Een database en tabel maken om de Twitter-data op te slaan

Met zowel toegang tot de Twitter API als SQLite geïnstalleerd, kunnen we eindelijk beginnen met het bouwen van een pijplijn om tweets op te slaan terwijl we ze in de loop van de tijd streamen. We beginnen met het aanmaken van onze nieuwe SQLite-database met R zoals hieronder beschreven:

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

Vervolgens kunnen we een tabel in de database aanmaken om de tweets te bewaren. In ons specifieke geval slaan we de volgende variabelen op:

Tweet_ID als INTEGER primary key
User als TEXT
Tweet_Content als TEXT
Date_Created als INTEGER

Je vraagt je misschien af: waarom sla ik de datums als integers op? Dit komt doordat SQLite geen gereserveerd datatype heeft voor datums en tijden. Daarom worden datums opgeslagen als het aantal seconden sinds 1970-01-01.

Laten we nu de tabel uitschrijven:

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

Nadat je de tabel hebt aangemaakt, kun je naar sqlite3.exe gaan en controleren of deze inderdaad is aangemaakt. Hieronder zie je een screenshot:

Stap 2: Stream tweets over je favoriete onderwerpen!

Geloof het of niet: qua vereisten en infrastructuur ben je klaar om een eenvoudige, functionele Twitter-streamingpijplijn te hebben. Wat we nu moeten doen, is tweets streamen via de API. Ter info: voor deze tutorial gebruik ik de standaard gratis API. Er zijn betaalde premiumversies die wellicht beter aansluiten op je behoeften als je bijvoorbeeld onderzoek doet.

Zonder verdere omhaal starten we het instellen van onze Twitter-listener. Het eerste wat je nodig hebt, is het rtweet-pakket importeren en de access tokens en secrets van je applicatie invoeren zoals aan het begin beschreven:

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

Met je token gereed is de volgende stap bepalen welke tweets je wilt streamen (oftewel: beluisteren). De functie stream_tweets in het rtweet-pakket biedt allerlei opties om de Twitter API te bevragen. Zo kun je tweets streamen die één of meer van een reeks opgegeven hashtags of keywords bevatten (tot 400), een kleine willekeurige subset van alle publiek beschikbare tweets, de tweets volgen van een groep user-ID’s of schermnamen (tot 5000), of tweets verzamelen op basis van geografische locatie.

Voor deze tutorial heb ik ervoor gekozen om tweets te streamen die hashtags rond data science bevatten (zie onderstaande lijst). Het valt je misschien op dat de manier waarop ik de hashtags heb gedefinieerd wat vreemd oogt. Maar dat is het vereiste formaat voor de functie stream_tweets wanneer je hashtags of keywords wilt beluisteren. Dit formaat verschilt als je een bepaalde set gebruikers wilt volgen of op coördinaten wilt filteren. Zie voor meer details de documentatie.

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

Met de keywords gedefinieerd is het tijd om de streaminglus op te zetten. Er zijn meerdere manieren om dit te doen, maar dit format heeft voor mij in het verleden goed gewerkt:

# 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 wezen streamt die lus zo veel mogelijk tweets die een van de hashtags in de key-string noemen, in intervallen van 2 uur voor in totaal 12 uur. Om de 2 uur maakt de Twitter-listener een .json-bestand aan in je huidige werkdirectory met de naam die is opgegeven in de variabele filename.

Vervolgens geeft hij deze bestandsnaam door aan de functie transform_and_clean_tweets, die indien gewenst retweets verwijdert, de kolommen selecteert die we willen behouden uit alles wat de Twitter API teruggeeft, en de tekst in de tweets normaliseert.

Daarna voegt hij de resulterende dataframe toe aan de tabel Tweet_Data die we eerder in onze SQLite-database hebben aangemaakt. Ten slotte telt hij 2 uur op bij de uurenteller (aangezien de streams 2 uur duren) en verwijdert hij het aangemaakte .json-bestand. Dit doen we omdat alle relevante data nu in onze database staat, en het bewaren van de .json-bestanden een opslaglast kan worden.

Laten we nu de functie transform_and_clean_tweets van dichtbij bekijken:

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

Zoals opgemerkt is deze functie bedoeld om indien gewenst retweets te filteren, de gewenste features te behouden en de tweet-tekst te normaliseren. In essentie kun je dit zien als het "T"-gedeelte van het ETL-acroniem (de transformatie). Een belangrijk onderdeel van deze functie is het opschonen van de tweet-tekst.

Tekstdata vereist meestal een paar voorbewerkingsstappen voordat deze klaar is voor welke analyse dan ook. In het geval van tweets kunnen deze stappen het verwijderen van URL’s, stopwoorden en mentions omvatten, de tekst naar kleine letters omzetten, woorden stemmen, enzovoort. Niet al deze stappen zijn echter altijd nodig. Hoe dan ook, hieronder laat ik de functie normalize_text zien die ik gebruikte om deze tweets te preprocessen:

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

Afhankelijk van je use case kunnen deze stappen voldoende zijn. Zoals eerder genoemd kun je ervoor kiezen om meer of andere stappen toe te voegen, zoals stemming of lemmatisering, of om alleen letters te behouden in plaats van letters en cijfers. Ik moedig je aan om te experimenteren en verschillende combinaties te proberen. Dit is een uitstekende plek om je regex-vaardigheden te oefenen.

Na al deze stappen is het resultaat een SQLite-database die gevuld is met alle gestreamde tweets. Je kunt controleren of alles goed heeft gewerkt door een paar eenvoudige queries uit te voeren, bijvoorbeeld:

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-database gevuld met alle gestreamde tweets

print(as.numeric(unique_rows))

## [1] 1863

Stap 3: Analyseren

Als je zeker weet dat het ETL-proces werkt zoals bedoeld, is de laatste stap om inzichten te verzamelen en de data die je hebt opgehaald te analyseren. Voor de tweets die we hebben verzameld kunnen we bijvoorbeeld een paar dingen proberen: een wordcloud van termen die in de inhoud van de tweet worden genoemd en een tijdlijn om te visualiseren op welk moment in onze 12 uur durende streamingperiode we de meeste tweets hebben opgehaald. Deze lijst is uiteraard geen volledig overzicht van al het onderzoek dat je met tweets kunt doen; dat varieert van sentimentanalyse tot psychografische studies en meer.

Dat gezegd hebbende: laten we meteen een mooie wordcloud bouwen:

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

Nou, wat een enorme verrassing! machinelearning en datascience zijn de meest genoemde woorden in al onze tweets—toevallig precies twee van de hashtags waar we op streamden. Dat resultaat was dus verwacht. De andere woorden zijn echter leuker om naar te kijken. Zo waren bigdata en artificialintelligence geen hashtags in onze keys, maar ze kwamen wel vaak voor; je zou dus kunnen stellen dat er vaak over gesproken wordt samen met de andere twee. We pikten ook andere interessante woorden op, zoals python of tensorflow, die ons net wat meer context geven over de inhoud van de verzamelde tweets dan alleen de hashtags.

Laten we nu een andere eenvoudige analyse doen. Tijdens onze 12 uur durende stream: op welk tijdstip verzamelden we de meeste tweets? Daarvoor halen we onze integerdatums op, zetten we ze om naar het juiste formaat en plotten we daarna het aantal tweets in de tijd:

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

Mooi! Nu zien we dat het grootste aantal unieke tweets dat we binnenkregen afkomstig was van 2 september tussen 20:00 en 20:59 uur UTC (dit verschijnt in 24-uursnotatie als 20).

Conclusie

Gefeliciteerd! Je weet nu hoe je een eenvoudige ETL-pijplijn in R bouwt. De twee analyses die we hebben gedaan, zijn heel basale analyses met Twitter-data. Maar zoals eerder genoemd is er heel veel mogelijk, zolang je een robuuste pijplijn bouwt om de data binnen te halen. Dat aspect stond centraal in deze tutorial.

Dat gezegd hebbende: deze tutorial liet slechts een kleinschalige casestudy zien om het proces van ETL-pijplijnen voor Twitter-data te doorlopen. Het bouwen van robuuste en schaalbare ETL-pijplijnen voor een hele organisatie is een complexe onderneming die veel rekenkracht en kennis vereist, zeker wanneer big data in het spel is.

Ik moedig je aan om verder onderzoek te doen en je eigen kleinschalige pijplijnen te bouwen, bijvoorbeeld in Python. Ga misschien zelfs aan de slag met big-dataprojecten. DataCamp heeft bijvoorbeeld al cursussen zoals de Big Data Fundamentals via PySpark-cursus, waarin ze big data behandelen met tools als PySpark om je kennis in dit vakgebied te verdiepen.

Wil je meer leren over data engineering? Volg dan DataCamps cursus Introduction to Data Engineering. En als je klaar bent om je nieuwe vaardigheden aan werkgevers te laten zien, bekijk dan onze Data Engineer Certification.

Referenties