Un pipeline ETL (extraire, transformer, charger) est un type de flux de travail fondamental dans l'ingénierie des données. L'objectif est de prendre des données qui pourraient être non structurées ou difficiles à utiliser ou à accéder et de servir une source de données propres et structurées.

Il est également très simple et facile de créer un pipeline simple en tant que script Python. Le code source complet de cet exercice est ici.

Qu'est-ce qu'un pipeline ETL ?

Un pipeline ETL se compose de trois composants généraux :

Extraire - obtenir des données à partir d'une source telle qu'une API. Dans cet exercice, nous n'extraireons les données qu'une seule fois pour montrer comment c'est fait. Dans de nombreux cas, nous devrons interroger l'API à intervalles réguliers pour obtenir de nouvelles données (appelées batch), ce que nous faisons en créant des workflows ETL planifiés. Nous pouvons également travailler avec des sources de données en continu (continues) que nous gérons via des systèmes de messagerie.

Transformer — structurer, formater ou nettoyer les données, en fonction de la raison pour laquelle nous en avons besoin et de la manière dont elles doivent être livrées. La transformation des données peut se produire à de nombreuses étapes du cycle de vie des données, et différents utilisateurs peuvent avoir besoin de les transformer pour différentes raisons. Dans cet exercice, nous allons formater les données des réponses JSON aux dataframes pandas et les écrire au format CSV.

Charger— écrire les données dans une destination externe où elles peuvent être utilisées par une autre application. Dans cet exercice, nous allons écrire chaque tableau que nous créons au format CSV. Dans un pipeline de données réel, nous écrivions dans des bases de données ou d'autres magasins de données, stockions nos fichiers dans différents formats ou envoyions des données à différents endroits dans le monde.

Commencer

Nous allons demander des enregistrements à l'API The Movie Database. Avant de commencer, vous devrez obtenir une clé API pour faire des demandes. Vous pouvez trouver les instructions pour obtenir une clé ici.

Une fois que vous avez votre clé, vous devrez vous assurer de ne pas la mettre directement dans votre code source. Un moyen rapide de le faire est de créer un fichier appelé config.py dans le même répertoire que vous allez créer votre script ETL. Mettez ceci dans le fichier :

#config.py
api_key = <YOUR API KEY HERE>

Si vous publiez votre code n'importe où, vous devez mettre votre config.py dans un fichier .gitignore ou similaire pour s'assurer qu'il n'est pas poussé vers des référentiels distants. Vous pouvez également stocker la clé API en tant que variable d'environnement ou utiliser une autre méthode pour la masquer. L'objectif est de le protéger contre l'exposition dans le script ETL.

Créez maintenant un fichier appelé tmdb.py et importer les éléments nécessaires.

import pandas as pd
import requests
import config

Extraire

Voici comment nous envoyons une seule requête GET à l'API. Dans la réponse, nous recevons un enregistrement JSON avec le movie_id nous précisons :

API_KEY = config.api_key
url = 'https://api.themoviedb.org/3/movie/{}?api_key={}'.format(movie_id, API_KEY)

r = requests.get(url)

Pour cet exercice, nous allons demander 6 films avec movie_id allant de 550 à 555. Nous créons une boucle qui demande chaque film un par un et ajoute la réponse à une liste.

response_list = []
API_KEY = config.api_key

for movie_id in range(550,556): 
  url = 'https://api.themoviedb.org/3/movie/{}?api_key={}'.format(movie_id, API_KEY)
  r = requests.get(url)
  response_list.append(r.json())

Nous avons maintenant une liste d'enregistrements JSON longs et peu maniables qui nous sont livrés à partir de l'API. Créez une base de données pandas à partir des enregistrements à l'aide de from_dict():

df = pd.DataFrame.from_dict(response_list)

Si tout se passe bien, vous devriez avoir un dataframe proprement formaté avec 6 lignes et 38 colonnes.

Transformer

Nous n'allons pas travailler avec toutes ces colonnes dans cet ensemble de données, alors sélectionnons celles que nous allons utiliser. Voici ceux qui nous intéressent pour cet exercice :

• budget
• id
• imdb_id
• genres
• original_title
• release_date
• revenue
• runtime

Nous créons une liste de noms de colonnes appelée df_columns qui nous permet de sélectionner les colonnes que nous voulons à partir de la base de données principale.

df_columns = ['budget', 'genres', 'id', 'imdb_id', 'original_title', 'release_date', 'revenue', 'runtime']

Travailler avec des colonnes catégorielles

Travailler avec des colonnes catégorielles genres la colonne ressemble à :

Il s'agit d'une colonne de listes d'enregistrements JSON, difficile à lire ou à comprendre rapidement dans ce format. Nous souhaitons développer cette colonne afin de pouvoir facilement voir et utiliser les enregistrements internes.

Une façon de procéder consiste à exploser la colonne de listes en colonnes catégorielles uniques. Cela se fait en créant une seule colonne pour chaque valeur catégorielle et en définissant la valeur de la ligne sur 1 si le film appartient à cette catégorie et sur 0 si ce n'est pas le cas. Nous en verrons les résultats sous peu.

Nous n'utiliserons que le name la propriété, pas le id. Nous pourrons accéder au id valeur dans un tableau séparé.

genres_list = df['genres'].tolist()
flat_list = [item for sublist in genres_list for item in sublist]

Nous voulons créer un tableau séparé pour les genres et une colonne de listes à exploser. Nous allons créer une colonne temporaire appelée genres_all comme une liste de listes de genres que nous pouvons ensuite développer dans une colonne distincte pour chaque genre.

result = []
for l in genres_list:
    r = []
    for d in l:
        r.append(d['name'])
    result.append(r)
df = df.assign(genres_all=result)

Voici où nous créons la table des genres :

df_genres = pd.DataFrame.from_records(flat_list).drop_duplicates()

Cela nous donne un tableau des propriétés de genre name et id. We attach the list of genre names onto our df_columns list as shown below and remove the original genres column from the list.

Les deux dernières lignes sont celles où nous utilisons le explode() et crosstab() fonctions pour créer les colonnes de genre et les joindre sur la table principale. Cela prend une colonne de listes et la transforme en un ensemble de colonnes de valeurs de fréquence.

df_columns = ['budget', 'id', 'imdb_id', 'original_title', 'release_date', 'revenue', 'runtime']
df_genre_columns = df_genres['name'].to_list()
df_columns.extend(df_genre_columns)

s = df['genres_all'].explode()
df = df.join(pd.crosstab(s.index, s))

Nous avons maintenant les colonnes suivantes dans notre table de films :

Remarquez les colonnes de genre à droite. Si un film appartient à un genre, la valeur de ligne est 1, et sinon, la valeur est 0. Il est maintenant facile pour nous de filtrer sur des genres spécifiques et de dire rapidement si un film appartient à un genre ou non.

Travailler avec les dates-heures

Enfin, nous allons développer la colonne datetime dans un tableau. Pandas a des fonctions intégrées pour extraire des parties spécifiques d'une date/heure. Remarquez que nous devons convertir le release_date colonne dans un datetime d'abord.

df['release_date'] = pd.to_datetime(df['release_date'])
df['day'] = df['release_date'].dt.day
df['month'] = df['release_date'].dt.month
df['year'] = df['release_date'].dt.year
df['day_of_week'] = df['release_date'].dt.day_name()
df_time_columns = ['id', 'release_date', 'day', 'month', 'year', 'day_of_week']

Voici la trame de données définie par ce code :

Charger

Nous avons fini par créer 3 tables pour le tmdb schéma que nous appellerons movies, genreset datetimes. Nous exportons nos tables en les écrivant dans un fichier. Cela créera 3 fichiers .csv dans le même répertoire que notre script.

df[df_columns].to_csv('tmdb_movies.csv', index=False)
df_genres.to_csv('tmdb_genres.csv', index=False)
df[df_time_columns].to_csv('tmdb_datetimes.csv', index=False)

C'est ça! Nous avons créé notre premier pipeline ETL. Nous avons construit un schéma structuré à partir d'une liste d'enregistrements JSON et transformé nos données dans un format propre et utilisable. Vous pouvez expérimenter de nombreuses façons différentes d'explorer et de structurer ces données, car nous n'avons fait qu'effleurer la surface ici et utilisé une petite partie de ce dont nous disposions.