Mise en place Projet de Data Enginnering
Un projet de Data Engineering vise à mettre en place l’infrastructure et les processus nécessaires pour permettre à une organisation de collecter, stocker, traiter et analyser efficacement ses données afin de répondre à un besoin et d’en tirer des informations précieuses pour la prise de décision et d’autres cas d’utilisation.
L’architecture ci dessus montre les étapes qui seront faites durant le projet. Les données sont extraites sur le site du datagouv.fr et ce sont les données des transactions immobilières en France. Le langage python qui sera utilisé pour la partie ETL (Extraction, Transformation, Chargement “Load”). Les données seront ensuite stockées dans une base de données Duckdb et certaines analyses seront faite avec du SQL et enfin power BI sera utilisé pour la partie Analytics avec la mise en place d’un Dashboard power BI.
DuckDB est une base de données relationnelle conçue pour offrir des performances élevées et une facilité d’utilisation dans les environnements interactifs et analytiques. Avec son architecture sans serveur, son traitement par colonnes et sa prise en charge du langage SQL, DuckDB est bien adapté aux applications nécessitant des analyses rapides sur de grands ensembles de données. Si vous souhaitez avoir plus d’information sur cette base de données je vous mets le lien :
Implémentation :
Pour commencer, il a fallu créer un repo git puis la préparation de l’environnement de travail.
Par la suite, il a fallu écrire le code qui permettrait d’extraire les données
| # Import the module import pandas as pd def extraction_data(filepath: object, select_col: list) -> object: """ This fonction extract data of real estate and take tree parameters : param filepath: str file txt : output: data extracted in csv format """ try: estate_df = pd.read_csv(filepath, sep = "|", low_memory = False) estate_df = estate_df[select_col] print(f"Data : {filepath} have extracted correctly.") except FileNotFoundError as e: print(f"Error, file {filepath} is not found.") except Exception as e: print(f"Error : {e} during extraction.") return estate_df def extract_data_dpt(filepath : object): """ This fonction extracte data of France department :param filepath : str file in csv data :output : data extracted from csv file """ try: dept_df = pd.read_csv(filepath) print(f"Data of {filepath} extracted correctly.") except Exception as e: print(f"Error : {e} during extraction.") return dept_df |
Une fois l’extraction terminée, le processus de transformation va commencer et le code permettant de réaliser cette transformation est :
# Import the module import pandas as pd col_rename = {"no disposition": "no_disposition", "date mutation": "date_mutation", "nature mutation": "nature_mutation", "valeur fonciere": "valeur_fonciere", "code voie": "code_voie", "code postal": "code_postal", "code departement": "code_departement", "code commune": "code_commune", "nombre de lots": "nombre_de_lots", "code type local": "code_type_local", "type local": "type_local", "surface reelle bati": "surface_reelle_bati", "nombre pieces principales": "nombre_pieces"} col_drop = ['No Volume', '1er lot', 'Surface Carrez du 1er lot', '2eme lot', 'Surface Carrez du 2eme lot', '3eme lot', 'Surface Carrez du 3eme lot', '4eme lot', 'Surface Carrez du 4eme lot', '5eme lot', 'Surface Carrez du 5eme lot'] def transform_data(estate_df : object): """ This function transform data which have extract. :param reat_estate: extracted data file in csv format :outpout: data file transformed """ # delete columns with more 90% missing values estate_df = estate_df.drop(col_drop, axis = 1) # convert columns to appropriate data types estate_df['Date mutation'] = pd.to_datetime(estate_df['Date mutation'], format ='%d/%m/%Y') estate_df['Valeur fonciere'] = estate_df['Valeur fonciere'].str.replace(',', '.') estate_df['Valeur fonciere'] = estate_df['Valeur fonciere'].astype(float) estate_df.columns = estate_df.columns.str.lower() # rename columns estate_df = estate_df.rename(columns = col_rename) print("The process is transformation is finish") return estate_df |
Une fois les données transformées correctement, elles seront donc stockées dans la base de données duckdb. On crée une fonction qui nous permet de se connecter à cette base et charger les donnée
|
import duckdb # create the connexion to a database (our database is file that we called real_estate.db) def load_data(estate_df, table_name): print("Data is load correctly")
|
xxxxx
Exemple d’ Analyse avec SQL:
On peut chercher à voir le nombre de vente par nature de mutation
| con.execute( """ SELECT nature_mutation, COUNT(nature_mutation) AS nbr_ventes FROM immo_france GROUP BY nature_mutation """ ).df() |
x
On peut calculer certaines statistiques par exemple Calcul de la moyenne (moyenne mobile) des 7 derniers jours des valeur_foncière dans les Yvelines. On va utiliser (Les fonction de fenêtrage (les Windows fonction) : consistent à prendre des bloques de données pour calculer les statistiques sur un bloc précis de la dataframe).
"""
Calcul de la moyenne (moyenne mobile) des 7 derniers jours des valeur_foncière dans les Yvelines
"""
con.execute(
"""
SELECT
date_mutation,
DAYOFWEEK(date_mutation) AS jour_semaine,
MONTH(date_mutation) AS numero_mois,
nature_mutation,
valeur_fonciere,
code_departement,
nom_departement,
nom_region,
CAST(AVG(valeur_fonciere) OVER(
PARTITION BY jour_semaine
ORDER BY date_mutation
ROWS BETWEEN 6 PRECEDING
AND CURRENT ROW
) AS INTEGER) AS mean_3last_month
FROM immo_france
LEFT JOIN zone_geographique
USING(code_departement)
WHERE (nom_departement = 'Yvelines' AND jour_semaine = 6 AND numero_mois = 1)
"""
).df().head(10)
|
-
ANALYTICS (DASHBORD):
Une fois que les données sont stockées dans la base on peut mettre en place un tableau de bord power BI.
Modèle de données:
