ETL/ELT con panda

Per l'utilizzo con i SKOOR Dashboard, i dati possono essere letti da varie fonti, quali database, API REST o file, e caricati in un database locale. Se necessario, i dati vengono trasformati prima o dopo il caricamento. Questa elaborazione, denominata ETL (extract, transform, load) o ELT (extract, load, transform), viene eseguita dai cosiddetti convertitori presenti nella soluzione SKOOR. I convertitori SKOOR vengono in genere creati utilizzando Talend o Pandas. Seguendo una struttura di directory specifica, possono essere compilati e compressi in un file ZIP per essere caricati nella sezione Importazione dati. Questa guida illustra la configurazione di base di un convertitore Pandas.

Si prega di visitare il sito web della documentazione di Pandas per i dettagli sul prodotto, il riferimento ai comandi, ecc.

Prerequisiti

I convertitori utilizzano in genere i seguenti moduli Python:

pandas (il software pandas)
pyarrow (una libreria utile)
openpyxl (libreria per leggere/scrivere file Microsoft Excel)
sqlalchemy (libreria per accedere a vari database)
psycopg2-binary (adattatore per il database PostgreSQL)

Esempio di convertitore

Configurazione di base

Intestazione dello 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)

Lettura di un file Excel e scrittura nel database

La parte seguente dello script può essere utilizzata come punto di partenza per l'elaborazione di un file Excel e può essere aggiunta sotto l'intestazione dello script sopra riportata:

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

Inserimento o aggiornamento dei dati in PostgreSQL

La funzione to_sql di pandas può essere utilizzata per inserire o sostituire dati. Se i dati devono essere aggiornati, è possibile utilizzare la libreria SQLAlchemy. Il seguente frammento di codice mostra come ottenere questo risultato:

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

Creazione di un convertitore

In ogni directory del convertitore, il servizio ETL SKOOR cerca uno script denominato <nome-directory>_run.sh che verrà eseguito all’avvio del convertitore. Creare una directory denominata <nome-directory> contenente questo script e lo script Python con la logica di elaborazione:

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

Esempio di script customers_run.sh:

#!/bin/bash

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

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

Ora è necessario creare un file zip contenente la directory del convertitore con gli script. Questo file zip può essere caricato nella sezione Caricamento dati delle SKOOR Dashboard.

Creare un processo SKOOR per automatizzare un convertitore

Se necessario, i convertitori possono essere eseguiti automaticamente tramite un job. Configurare un job di esecuzione per eseguire lo script *_run.sh sopra indicato nella struttura di directory di SKOOR ETL:

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

Esempio:

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