FI-DPA 04 ETL-ELT-Strecken

FI-DPA 04 ETL/ELT-Strecken

ETL/ELT-Prozesse bilden das Rückgrat moderner Datenintegration. In diesem Modul erlernen Sie die Konzepte zur Extraktion, Transformation und Laden von Daten zwischen verschiedenen Systemen. Sie verstehen, wie Staging-Areas zur Qualitätsicherung eingesetzt werden und warum Idempotenz in der Datenverarbeitung entscheidend ist. Praktische Beispiele mit Airflow, dbt und Talend runden Ihr Wissen ab.

Sie werden in der Lage sein, ETL/ELT-Pipelines zu entwerfen, zu implementieren und zu warten. Der Fokus liegt auf robusten, wartbaren und performanten Lösungen für den Einsatz in KMU-Umgebungen.

Konzepte und Hintergrund

ETL (Extract, Transform, Load)

Der klassische Ansatz, bei dem Daten zuerst extrahiert, dann in einer separaten Datenbank transformiert und schließlich in das Zielsystem geladen werden. Die Transformation erfolgt vor dem Laden in das Zielsystem.

ELT (Extract, Load, Transform)

Ein moderner Ansatz, bei dem Daten direkt in das Zielsystem (meist ein Data Warehouse) geladen und dort transformiert werden. Dies nutzt die Skalierbarkeit moderner Data-Warehouses und ermöglicht flexiblere Transformationen.

Staging-Area

Ein temporärer Speicherbereich, der als Zwischenstation für Daten während des ETL-Prozesses dient. Hier werden Daten bereinigt, validiert und aufbereitet, bevor sie in das Zielsystem geladen werden.

Idempotenz

Eine Eigenschaft von Operationen, die mehrmals ausgeführt werden können, ohne das Ergebnis zu verändern. Im ETL-Kontext bedeutet dies, dass die Ausführung einer Pipeline mehrmals zum identischen Ergebnis führt, auch wenn Datenquellen bereits verarbeitet wurden.

Orchestrierung

Die Koordination verschiedener Tasks und Workflows in einem ETL-Prozess. Tools wie Airflow ermöglichen die Planung, Ausführung und Überwachung komplexer Datenpipelines.

Architektur-Diagramm

flowchart LR
    A[Datenquellen] --> B(Extraktion)
    B --> C[Staging-Area]
    C --> D[Transformation]
    D --> E[Zielsystem]
    F[Airflow] --> B
    F --> D
    G[dbt] --> D
    H[Talend] --> B
    H --> D

Praktische Schritte

1. Installieren Sie Apache Airflow mit pip in einer virtuellen Umgebung. Dies stellt eine saubere Trennung von Projekten sicher.

python -m venv airflow-env
source airflow-env/bin/activate
pip install apache-airflow

2. Konfigurieren Sie die Airflow-Datenbank mit PostgreSQL für verbesserte Performance und Skalierbarkeit.

export AIRFLOW__CORE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:password@localhost/airflow
airflow db init

3. Erstellen Sie in dbt ein neues Projekt und konfigurieren Sie die Verbindungsparameter zu Ihrer Staging-Datenbank.

dbt init my_project
cd my_project
vim profiles.yml

4. Definieren Sie in dbt Modelle für Ihre Transformationen mit SQL-Dateien im Ordner models/.

mkdir -p models/staging
cat > models/staging/customers.sql <<EOF
select id, name, email from raw.customers
where email is not null
EOF</code>

5. Erstellen Sie in Airflow einen DAG, der dbt aufruft, nachdem die Daten extrahiert wurden.

from airflow.operators.bash import BashOperator
from airflow import DAG
from datetime import datetime

with DAG('etl_pipeline', start_date=datetime(2023,1,1)) as dag:
    run_dbt = BashOperator(
        task_id='run_dbt',
        bash_command='dbt run'
    )

6. Implementieren Sie in Talend eine Job-Komponente zur Datenbereinigung, die doppelte Einträge basierend auf einer eindeutigen ID entfernt.

tRowGenerator --component tUniqueRow --component tFilterRow --component tLogRow

7. Konfigurieren Sie in Airflow Retry-Policies für kritische Tasks, um vorübergehende Fehler zu handhaben.

run_dbt = BashOperator(
    task_id='run_dbt',
    bash_command='dbt run',
    retries=3,
    retry_delay=timedelta(minutes=5)
)

8. Implementieren Sie in dbt Tests zur Sicherstellung der Datenqualität, z.B. Eindeutigkeit von IDs.

cat > models/staging/schema.yml <<EOF
version: 2

models:
  - name: customers
    columns:
      - name: id
        tests:
          - unique
EOF</code>

9. Verwenden Sie in Airflow den Sensor-Operator, um auf das Erscheinen neuer Dateien in einem Verzeichnis zu warten.

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id='wait_for_file',
    filepath='/data/incoming/*.csv',
    poke_interval=60,
    timeout=3600
)

10. Konfigurieren Sie in Talend eine Verbindung zu Ihrer Datenbank mit Verbindungspooling für verbesserte Performance.

tDBConnection --component tDBOutput --component tDBCommit

-- models/marts/fct_sales.sql
{{ config(
    materialized='incremental',
    unique_key='order_id'
) }}

-- Wir lesen aus der staging-Tabelle
WITH source_data AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

-- Nur neue oder geänderte Datensätze auswählen (Delta)
new_records AS (
    SELECT * FROM source_data
    WHERE order_date >= (SELECT COALESCE(MAX(order_date), '1900-01-01') FROM {{ this }})
)

-- Idempotenter Merge: Aktualisiert existierende, fügt neue hinzu
MERGE INTO {{ this }} AS target
USING new_records AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET
    target.amount = source.amount,
    target.status = source.status,
    target.updated_at = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN INSERT
    (order_id, amount, status, updated_at)
    VALUES
    (source.order_id, source.amount, source.status, CURRENT_TIMESTAMP());

# dags/elt_logistics.py
from airflow import DAG
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudExecutionTask
from airflow.providers.dbt.cloud.triggers.dbt import DbtCloudTrigger

# ... (DAG Definition und Python-Operator für das Loading)

# Schritt 1: Loading nach Raw
load_raw = PythonOperator(
    task_id='load_raw_orders',
    python_callable=load_csv_to_snowflake,
)

# Schritt 2: dbt Run (Transformation)
# dbt Cloud Trigger erlaubt asynchrones Warten auf den Job
transform_dbt = DbtCloudTriggerOperator(
    task_id='run_dbt_incremental',
    dbt_cloud_conn_id='dbt_cloud_conn',
    trigger_id=12345, # ID des dbt Cloud Jobs
    wait=True,
)

load_raw >> transform_dbt