ETL/ELT avec pandas

Pour une utilisation avec les SKOOR Dashboards, les données peuvent être extraites de diverses sources, telles que des bases de données, des API REST ou des fichiers, puis chargées dans une base de données locale. Si nécessaire, elles sont transformées avant ou après leur chargement. Ce traitement, appelé ETL (extraction, transformation, chargement) ou ELT (extraction, chargement, transformation), est effectué par des « convertisseurs » intégrés à la solution SKOOR. Les convertisseurs SKOOR sont généralement créés à l’aide de Talend ou de Pandas. En respectant une structure de répertoires spécifique, ils peuvent être compilés et compressés pour être téléchargés dans la section « Importation de données ». Ce guide présente la configuration de base d’un convertisseur Pandas.

Veuillez consulter le site de documentation de Pandas pour plus de détails sur le produit, le référentiel des commandes, etc.

Prérequis

Les modules Python suivants sont généralement utilisés par les convertisseurs :

pandas (le logiciel pandas)
pyarrow (une bibliothèque utile)
openpyxl (bibliothèque permettant de lire et d'écrire des fichiers Microsoft Excel)
sqlalchemy (bibliothèque permettant d'accéder à diverses bases de données)
psycopg2-binary (adaptateur de base de données PostgreSQL)

Exemple de convertisseur

Configuration de base

En-tête du script Pandas :

import pandas as pd
from sqlalchemy import create_engine
import numpy as np
import sys, getopt
import datetime

# Variables
# =========================================================================================

# local PostgreSQL database to load data
pgUser = '<db user>'
pgPass = '<db password>'
pgDatabase = '<database name>'
pgHost = 'localhost'    # Change this host if necessary
pgPort = 5432           # Change this port if necessary

# Functions
# =========================================================================================

# process SKOOR ETL service args
def getArg(argument):
   try:
      opts, args = getopt.getopt(sys.argv,"",["sourceFile=", "sessionId="])
      for arg in args[1:]:
          splitArg = arg.split('=')
          if splitArg[0] == argument:
              return splitArg[1]

   except getopt.GetoptError:
      print(args[0], 'sourceFile=<source.file>')
      sys.exit(1)

Lecture d'un fichier Excel et écriture dans la base de données

La partie suivante du script peut servir de point de départ pour le traitement d’un fichier Excel et peut être ajoutée sous l’en-tête du script ci-dessus :

# Main script
# =========================================================================================
source_file = getArg("sourceFile")

# Check if a filename argument is available
if isinstance(source_file, str):
   print("Processing file " + source_file)

else:
   print("No input file defined!")
   exit(1)

# Create SqlAlchemy engine
engine = create_engine('postgresql://' + pgUser + ':' + pgPass + '@' + pgHost + ':' + str(pgPort) + '/' + pgDatabase)

# Read raw data from a sheet called "customer"
customers = pd.read_excel(source_file, engine='openpyxl', sheet_name='customer')

# Transform data if required
# ------------------------------------------------------------------------

<transformation code>

print(customers.head())

# Write data into table "customers" of PostgreSQL database
# ------------------------------------------------------------------------
print("Writing data to database...")
customers.to_sql(name='customers', con=engine, if_exists='replace')

print("Done")

Insérer ou mettre à jour des données dans PostgreSQL

La fonction to_sql de pandas peut être utilisée pour insérer ou remplacer des données. Si les données doivent être mises à jour, la bibliothèque SQLAlchemy peut être utilisée. L'extrait de code suivant montre comment procéder :

# Write DataFrame to temporary table on database
customers.to_sql(name='customers_temp', con=engine, if_exists='replace')

# Update target table using temporary table values
update_sql = '''INSERT INTO customers (customer_id, customer_name, customer_address)
    SELECT customer_id, customer_name, customer_address FROM customers_temp t
    ON CONFLICT (customer_id)
    DO UPDATE SET 
        customer_name = EXCLUDED.customer_name,
        customer_address = EXCLUDED.customer_address'''

with engine.begin() as conn:
    conn.execute(text(update_sql))

Créer un convertisseur

Dans chaque répertoire de convertisseur, le service ETL SKOOR recherche un script nommé <nom-du-répertoire>_run.sh qui sera exécuté lors du lancement du convertisseur. Créez un répertoire nommé <nom-du-répertoire> contenant ce script ainsi que le script Python contenant la logique de traitement :

$ ls -1 customers/
customers.py
customers_run.sh

Exemple de script customers_run.sh :

#!/bin/bash

cd $(dirname $0)
home=$(pwd)

/opt/eranger/python3-env/bin/python3 $home/customers.py $@

Il faut ensuite créer un fichier zip contenant le répertoire du convertisseur et ses scripts. Ce fichier zip peut être chargé dans la section « Chargement des données » des SKOOR Dashboard.

Créer une tâche SKOOR pour automatiser un convertisseur

Si nécessaire, les convertisseurs peuvent être exécutés automatiquement par une tâche. Configurez une tâche d’exécution pour exécuter le script *_run.sh ci-dessus dans la structure de répertoires de SKOOR ETL :

/var/opt/run/eranger/eranger-etl/converters/<converter>/<converter>/<converter>_run.sh

Exemple :

/var/opt/run/eranger/eranger-etl/converters/customers/customers/customers_run.sh