ETL/ELT mit Pandas

Für die Verwendung mit SKOOR Dashboards können Daten aus verschiedenen Quellen wie Datenbanken, REST-APIs oder Dateien ausgelesen und in eine lokale Datenbank geladen werden. Bei Bedarf werden diese Daten entweder vor oder nach dem Laden transformiert. Diese Verarbeitung, die als ETL (Extract, Transform, Load) oder ELT (Extract, Load, Transform) bezeichnet wird, erfolgt durch sogenannte Konverter in der SKOOR-Lösung. SKOOR-Konverter werden in der Regel mit Talend oder Pandas erstellt. Unter Beachtung einer bestimmten Verzeichnisstruktur können sie erstellt und als ZIP-Datei komprimiert werden, um sie im Bereich „Datenimport“ hochzuladen. Diese Anleitung behandelt die grundlegende Einrichtung eines Pandas-Konverters.

Weitere Informationen zum Produkt, zur Befehlsreferenz usw. finden Sie auf der Pandas-Dokumentationswebsite.

Voraussetzungen

Die folgenden Python-Module werden in der Regel von Konvertern verwendet:

pandas (die Pandas-Software)
pyarrow (eine hilfreiche Bibliothek)
openpyxl (Bibliothek zum Lesen und Schreiben von Microsoft-Excel-Dateien)
sqlalchemy (Bibliothek für den Zugriff auf verschiedene Datenbanken)
psycopg2-binary (PostgreSQL-Datenbankadapter)

Beispiel-Konverter

Grundlegende Einrichtung

Pandas-Skript-Header:

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)

Excel-Datei lesen und in die Datenbank schreiben

Der folgende Teil des Skripts kann als Ausgangspunkt für die Verarbeitung einer Excel-Datei dienen und unterhalb des obigen Skript-Headers eingefügt werden:

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

Daten in PostgreSQL einfügen oder aktualisieren

Die pandas-Funktion `to_sql` kann zum Einfügen oder Ersetzen von Daten verwendet werden. Wenn Daten aktualisiert werden müssen, kann die SQLAlchemy-Bibliothek verwendet werden. Der folgende Codeausschnitt zeigt, wie dies erreicht werden kann:

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

Konverter erstellen

In jedem Konverter-Verzeichnis sucht der SKOOR-ETL-Dienst nach einem Skript namens <Verzeichnisname>_run.sh, das beim Ausführen des Konverters ausgeführt wird. Erstellen Sie ein Verzeichnis namens <Verzeichnisname> mit diesem Skript darin und dem Python-Skript, das die Verarbeitungslogik enthält:

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

Beispielskript customers_run.sh:

#!/bin/bash

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

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

Nun muss eine ZIP-Datei erstellt werden, die das Konverter-Verzeichnis mit den Skripten enthält. Diese ZIP-Datei kann im Abschnitt „Datenladen“ des SKOOR Dashboard hochgeladen werden.

Erstellen Sie einen SKOOR-Job, um einen Konverter zu automatisieren

Bei Bedarf können Konverter automatisch über einen Job ausgeführt werden. Konfigurieren Sie einen Ausführungsjob, um das oben genannte *_run.sh-Skript in der Verzeichnisstruktur von SKOOR ETL auszuführen:

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

Beispiel:

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