Créer des pipelines de données avec R

Apprenez à créer et charger (ETL) un pipeline de données de zéro avec R et SQLite pour collecter des tweets en temps réel et les stocker pour de futures analyses.

Actualisé 19 sept. 2026 · 13 min lire

Explorer avec l’IA

ChatGPT Claude Perplexity

On peut dire que des pipelines ETL bien conçus sont un organe vital de la data science. Sans données propres et structurées, il devient difficile de produire des insights de qualité qui éclairent les décisions métiers.

Dans ce tutoriel, nous allons donc voir comment construire un pipeline ETL simple pour diffuser des Tweets en temps réel directement vers une base SQLite avec R. C’est une tâche assez courante en analyse des réseaux sociaux, par exemple.

Nous mettrons l’accent sur la démarche de collecte et de stockage des données ainsi que sur l’utilisation de l’API Twitter avec le package rtweet dans R.

Pour commencer, assurons-nous d’avoir les bons outils. La première étape consiste à configurer votre accès à l’API Twitter. Globalement, suivez les étapes ci-dessous :

Créez un compte Twitter si vous n’en avez pas.
Suivez ce lien et demandez un compte développeur (notez que ce processus nécessite désormais une demande soumise à l’approbation de Twitter).
Créez une nouvelle application sur cette page.
Renseignez tous les détails de votre application et générez votre jeton d’accès.
Récupérez votre consumer key, consumer secret, access token et access token secret, nécessaires pour vous connecter à l’API.

Une fois votre accès à l’API Twitter prêt, installez SQLite si besoin. Pour connaître l’intégralité de la procédure d’installation sur votre ordinateur, consultez le tutoriel Guide du débutant sur SQLite sur DataCamp. Nous avons choisi SQLite ici pour sa simplicité d’utilisation.

Étape 1 : créer une base et une table pour stocker les données Twitter

Avec l’accès à l’API Twitter et SQLite installés, nous pouvons commencer à construire un pipeline qui stocke les Tweets au fil du flux. Pour démarrer, créons une nouvelle base SQLite depuis R comme suit :

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

Ensuite, créons une table dans la base pour accueillir les tweets. Dans notre cas, nous allons stocker les variables suivantes :

Tweet_ID en INTEGER, clé primaire
User en TEXT
Tweet_Content en TEXT
Date_Created en INTEGER

Vous vous demandez peut-être pourquoi enregistrer les dates en entiers ? Parce que SQLite n’a pas de type dédié aux dates et heures. Les dates seront donc stockées en nombre de secondes depuis le 01/01/1970.

Passons maintenant à l’écriture de la table :

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

Une fois la table créée, vous pouvez ouvrir sqlite3.exe et vérifier qu’elle existe bien. Voir la capture d’écran ci-dessous :

Étape 2 : diffusez des Tweets sur vos sujets favoris !

Incroyable mais vrai : vous avez déjà tout ce qu’il faut, côté prérequis et infrastructure, pour un pipeline de streaming Twitter simple et fonctionnel. Il nous reste à diffuser des Tweets via l’API. À noter : pour ce tutoriel, j’utilise l’API standard gratuite. Des versions premium payantes peuvent mieux convenir à vos besoins de streaming, par exemple pour de la recherche.

Passons sans tarder à la configuration de notre écouteur Twitter. Commencez par importer le package rtweet et renseigner les jetons et secrets d’accès de votre application, comme indiqué au début :

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

Une fois le jeton en place, il faut décider quels tweets vous souhaitez diffuser (c’est-à-dire écouter). La fonction stream_tweets du package rtweet offre de nombreuses options pour interroger l’API Twitter. Par exemple, vous pouvez diffuser les tweets contenant un ou plusieurs hashtags ou mots-clés (jusqu’à 400), un petit sous-ensemble aléatoire de tous les tweets publics, suivre les tweets d’un groupe d’identifiants ou de noms d’utilisateur (jusqu’à 5 000), ou encore récupérer des tweets par zone géographique.

Pour ce tutoriel, j’ai choisi de diffuser les tweets contenant des hashtags liés à la data science (voir la liste ci-dessous). Vous remarquerez que le format de définition des hashtags à diffuser est un peu particulier. C’est néanmoins le format requis par stream_tweets lorsque vous écoutez des hashtags ou mots-clés. Le format change si vous écoutez un ensemble d’utilisateurs ou selon des coordonnées. Pour plus de détails, consultez la documentation.

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

Avec les mots-clés définis, il est temps de créer la boucle de streaming. Il existe plusieurs approches, mais celle-ci m’a très bien servi par le passé :

# 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 substance, cette boucle diffuse autant de tweets que possible mentionnant l’un des hashtags de la chaîne de clés, par intervalles de 2 heures pendant 12 heures au total. Toutes les 2 heures, l’écouteur Twitter crée un fichier .json dans votre répertoire de travail courant avec le nom défini dans la variable filename.

Ce nom de fichier est ensuite transmis à la fonction transform_and_clean_tweets qui supprime les retweets si souhaité, sélectionne les colonnes à conserver parmi celles renvoyées par l’API Twitter, et normalise le texte des Tweets.

Le dataframe résultant est ensuite ajouté à la table Tweet_Data que nous avons créée plus tôt dans notre base SQLite. Enfin, la boucle incrémente le compteur d’heures de 2 (puisque chaque flux dure 2 heures) et supprime le fichier .json créé. Nous supprimons ce fichier car les données utiles sont désormais dans la base, et conserver les .json peut vite poser un problème d’espace.

Regardons de plus près la fonction 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)
}

Comme indiqué, cette fonction filtre les retweets si nécessaire, conserve les variables souhaitées et normalise le texte des tweets. C’est, en somme, la partie « T » de l’acronyme ETL (transformation). Un élément clé est le nettoyage du texte.

En général, les données textuelles nécessitent quelques étapes de prétraitement avant toute analyse. Pour des tweets, cela peut inclure la suppression des URL, des mots vides et des mentions, la mise en minuscules, le stemming, etc. Cependant, toutes ces étapes ne sont pas toujours nécessaires. Voici, pour l’instant, la fonction normalize_text utilisée pour prétraiter ces 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)
}

Selon votre cas d’usage, ces étapes peuvent suffire. Comme mentionné plus haut, vous pouvez ajouter d’autres étapes (stemming, lemmatisation) ou conserver uniquement les lettres plutôt que lettres et chiffres. Expérimentez et testez différentes combinaisons : c’est un excellent terrain pour pratiquer les regex.

Après toutes ces étapes, vous obtenez une base SQLite remplie des tweets diffusés. Vous pouvez valider le bon fonctionnement avec quelques requêtes simples, par exemple :

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

Base SQLite remplie avec tous les tweets diffusés

print(as.numeric(unique_rows))

## [1] 1863

Étape 3 : analyser

Une fois certain que le processus ETL fonctionne comme prévu, la dernière étape consiste à extraire des insights et analyser les données collectées. Pour les tweets que nous avons rassemblés, testons par exemple deux pistes : un nuage de mots des termes mentionnés dans le contenu des tweets, et une frise temporelle pour visualiser à quel moment, sur la période de 12 heures, nous avons obtenu le plus de tweets. Cette liste est évidemment loin d’être exhaustive : on peut aller de l’analyse de sentiments à des études psychographiques, et bien plus.

Allons droit au but et construisons un joli nuage de mots :

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

Sans surprise ! machinelearning et datascience sont les mots les plus mentionnés dans nos tweets : ce sont deux des hashtags que nous avons diffusés. Résultat attendu. Les autres mots, en revanche, sont plus intéressants. Par exemple, bigdata et artificialintelligence n’étaient pas dans nos clés, mais apparaissent fréquemment : on peut en déduire qu’ils sont souvent cités avec les deux premiers. D’autres termes comme python ou tensorflow donnent davantage de contexte sur le contenu des tweets, au-delà des hashtags.

Passons à une autre analyse simple. Pendant nos 12 heures de diffusion, à quel moment avons-nous recueilli le plus de tweets ? Pour cela, récupérons nos dates stockées en entiers, convertissons-les au bon format, puis traçons la quantité de tweets dans le temps :

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

graphique de la quantité de tweets dans le temps

Parfait ! On voit maintenant que le plus grand nombre de tweets uniques provient du 2 septembre entre 20 h 00 et 20 h 59 UTC (affiché au format 24 h sous « 20 »).

Conclusion

Félicitations ! Vous savez désormais construire un pipeline ETL simple en R. Les deux analyses réalisées ici restent très basiques avec des données Twitter. Mais, comme indiqué, les possibilités sont nombreuses dès lors que vous mettez en place un pipeline robuste pour ingérer les données. C’était l’objectif principal de ce tutoriel.

Cela dit, ce tutoriel illustre un cas d’étude à très petite échelle pour parcourir les étapes de construction d’un pipeline ETL pour des données Twitter. Mettre en place des pipelines ETL robustes et scalables à l’échelle d’une entreprise est une tâche complexe qui requiert d’importantes ressources de calcul et des compétences avancées, en particulier avec du big data.

Je vous encourage à poursuivre vos recherches et à construire vos propres petits pipelines, par exemple en Python. Vous pouvez même vous lancer dans des projets big data. Par exemple, DataCamp propose déjà un cours comme Big Data Fundamentals via PySpark, qui couvre le big data avec des outils comme PySpark pour approfondir vos connaissances.

Si vous souhaitez vous initier au data engineering, suivez le cours Introduction to Data Engineering de DataCamp. Et si vous êtes prêt à prouver vos nouvelles compétences aux employeurs, découvrez notre Data Engineer Certification.

Références