Wprowadzenie do pythonowej biblioteki Data Load Tool (DLT)

Data Load Tool (DLT) to biblioteka open source dla Pythona, która umożliwia łatwe i skalowalne ładowanie danych z różnych źródeł do dobrze ustrukturyzowanych zbiorów danych. Dzięki swojej prostocie i elastyczności jest idealnym narzędziem do budowy procesów Extract and Load (EL), które mogą być zintegrowane z istniejącymi narzędziami, takimi jak DBT czy Airflow.

Data Load Tool (DLT) to biblioteka zaprojektowana z myślą o użytkownikach Pythona, którzy chcą zautomatyzować procesy ładowania danych. Oto kluczowe cechy tej technologii:

• obsługa różnych źródeł danych – REST API, bazy danych SQL, lokalne pliki (CSV, Parquet, JSONL) i wiele innych;
• automatyczne zarządzanie schematami – DLT automatycznie rozpoznaje schematy danych, obsługuje ich ewolucję oraz zapewnia spójność danych;
• elastyczność wdrożenia – można ją uruchamiać wszędzie tam, gdzie działa Python – od lokalnych środowisk po chmurę;
• bezproblemowa integracja – kompatybilność z popularnymi narzędziami, takimi jak Airflow, DBT czy funkcje serverless.

Architektura DLT

Biblioteka DLT składa się z kilku kluczowych komponentów:

• źródła danych (sources) – definiowane w kodzie Python, pozwalają na pobieranie danych z różnych systemów, np. API, baz danych czy plików;
• pipeline – określa, jak dane są przetwarzane i ładowane do systemu docelowego;
• destination (miejsce docelowe) – określa, gdzie dane zostaną załadowane, np. PostgreSQL, BigQuery, DuckDB;
• schematy i normalizacja – Data Load Tool automatycznie dostosowuje strukturę danych i zapewnia jej zgodność.

Zarządzanie schematami w Data Load Tool

Jedną z kluczowych funkcji DLT jest automatyczne zarządzanie schematami danych. Biblioteka:

• wykrywa zmiany w strukturze danych (np. dodanie nowych kolumn, zmiana typu danych);
• normalizuje dane i zapewnia ich zgodność z miejscem docelowym;
• obsługuje migracje schematów bez potrzeby ręcznej ingerencji użytkownika.

Dzięki temu Data Load Tool jest szczególnie przydatne w środowiskach, w których dane często zmieniają swoją strukturę.

Obsługiwane źródła i miejsca docelowe

Data Load Tool obsługuje szeroki zakres źródeł danych, w tym:

• bazy danych SQL (PostgreSQL, MySQL, SQLite, DuckDB, Snowflake, BigQuery, Redshift);
• REST API – możliwość pobierania danych z dynamicznych API z obsługą paginacji;
• lokalne i zdalne pliki – obsługa formatów CSV, JSONL czy Parquet;
• strumienie danych – wsparcie dla Kafka.

Systemy źródłowe, z których DLT może pobierać dane:

• relacyjne bazy danych, takie jak PostgreSQL i MySQL;
• hurtownie danych, np. BigQuery, Redshift, Snowflake;
• pliki w systemach plików lub w chmurze (np. S3, GCS).

Zarządzanie błędami i retry mechanism

Data Load Tool oferuje mechanizmy obsługi błędów, które pozwalają na stabilne działanie nawet w przypadku awarii.

1. Automatyczne ponawianie prób (retry mechanism) – w przypadku błędów sieciowych lub przeciążenia API, DLT automatycznie ponawia zapytania.
2. Logowanie i monitoring – biblioteka generuje czytelne logi, które ułatwiają debugowanie.
3. Walidacja danych – możliwość ustawienia reguł walidacyjnych, które sprawdzają poprawność danych przed ich załadowaniem.

Jak zacząć z Data Load Tool?

Aby rozpocząć pracę z DLT, należy zainstalować bibliotekę za pomocą pip:

bash
pip install dlt

W zależności od potrzeb można zainstalować Data Load Tool z obsługą konkretnej bazy danych, np:

bash
pip install "dlt[duckdb]"

bash
pip install "dlt[postgres]"

bash
pip install "dlt[filesystem]"

Zaleca się korzystanie z czystego środowiska wirtualnego dla eksperymentów. Po instalacji można zaimportować bibliotekę w skrypcie Pythona:

python
import dlt

Przykłady użycia DLT

Ładowanie danych z REST API do duckdb

Data Load Tool umożliwia łatwe pobieranie danych z API dzięki wbudowanemu źródłu REST API. Oto przykład konfiguracji:

python
from dlt.sources.rest_api import rest_api_source

source = rest_api_source({
    "client": {
        "base_url": "https://api.example.com/",
        "auth": {
            "token": dlt.secrets["your_api_token"],
        },
        "paginator": {
            "type": "json_link",
            "next_url_path": "paging.next",
        },
    },
    "resources": ["posts", "comments"],
})

pipeline = dlt.pipeline(
    pipeline_name="rest_api_example",
    destination="duckdb",
    dataset_name="rest_api_data",
)

load_info = pipeline.run(source)

Kod ten ilustruje proces konfiguracji klienta API, definiowania zasobów oraz uruchamiania potoku danych.

Ładowanie danych z bazy Oracle do bazy Postgres

Data Load Tool umożliwia łatwe ładowanie danych do bazy PostgreSQL, co jest przydatne w przypadku zarządzania dużymi zbiorami danych w relacyjnych bazach. Aby rozpocząć, należy zainstalować bibliotekę DLT z obsługą PostgreSQL:

bash
pip install "dlt[postgres]"

Następnie należy skonfigurować połączenie z bazą danych poprzez plik .dlt/secrets.toml:

[destination.postgres.credentials]
database = ""
username = ""
password = ""  # Uzupełnij hasło
host = "localhost"       # Lub adres IP serwera
port = 5432
connect_timeout = 15

Aby załadować dane z bazy danych, np. Oracle do bazy Postgres przy użyciu DLT, oprócz [destination] w pliku .dlt/secrets.toml, należy zadeklarować również [source], np.

[source.oracle.credentials]
database = "oracle_db"
username = "oracle_user"
password = ""  # Uzupełnij hasło
host = "oracle_host"     # Adres IP serwera Oracle
port = 1521

Poniższy przykład pokazuje, jak załadować dane z bazy Oracle do PostgreSQL:

import dlt
from dlt.sources.sql_database import sql_database

def load_select_tables_from_database():

    pipeline = dlt.pipeline(
    pipeline_name="oracle_to_postgres",
    destination="postgres",
    dataset_name="oracle_data"
 )
table = sql_table(table="family")
info = pipeline.run(table)
print(info)

Ładowanie danych z systemu plików do bazy Postgres

Data Load Tool wspiera również operacje na lokalnym systemie plików, umożliwiając ładowanie danych w formatach takich jak CSV, JSONL czy Parquet. Wystarczy zainstalować wersję biblioteki DLT obsługującą bazę Postgres:

bash
pip install "dlt[postgres]"

Tak jak w przykładzie powyżej, należy skonfigurować podłączenie do bazy Postgres w pliku .dlt/secrets.toml:

text
[destination.postgres.credentials]
database = ""
username = ""
password = ""  # Uzupełnij hasło
host = "localhost"       # Lub adres IP serwera
port = 5432
connect_timeout = 15

Poniżej przykład kodu dla ładowania danych z plików lokalnych do bazy danych.

import dlt
from dlt.sources.filesystem import filesystem, read_csv

filesystem_pipe = filesystem(bucket_url="file://Users/admin/Documents/csv_files", file_glob="*.csv") | read_csv()
pipeline = dlt.pipeline(pipeline_name="my_pipeline", 
  destination="postgres",
  dataset_name="filesystem_data")
info = pipeline.run(filesystem_pipe)
print(info)

Przykład użycia na podstawie wdrożenia systemu DORIAN dla WARTA – jednego z największych Towarzystw Ubezpieczeń i Reasekuracji w Polsce

Data Load Tool zostało wykorzystane m.in. w projekcie dla systemu DORIAN, gdzie pobierane są dane z systemów źródłowych klienta oraz ładowane do bazy docelowej, z której korzysta system. DORIAN to zaawansowane rozwiązanie wspierające kompleksowe zarządzanie ryzykiem operacyjnym w organizacjach objętych regulacjami DORA. System umożliwia identyfikację, ocenę i monitorowanie ryzyk w obszarze ICT, wspierając tym samym realizację wymagań dotyczących ciągłości działania, odporności cyfrowej oraz minimalizowania skutków incydentów operacyjnych. DORIAN integruje dane z różnych źródeł (m.in. ERP, CMDB, CRM), automatyzuje analizę zdarzeń, zarządza katalogiem ryzyk i pozwala na elastyczne wyznaczanie poziomu oraz wpływu ryzyka na podstawie konfigurowalnych reguł. Oferuje także intuicyjną wizualizację zależności, rozbudowane możliwości raportowania zgodne z regulacjami oraz precyzyjne zarządzanie dostępem dzięki modelowi RBAC.

Systemy źródłowe, z których DLT może pobierać dane:

• pliki .csv udostępniane przez https;
• pliki .json udostępniane przez podłączony zasób sieciowy;
• widoki w bazie danych Postgres;
• widoki w bazie danych Oracle;
• Rest-API Jira;
• Rest-API Jira-Insights.

Kluczowe zalety DLT

1. Prostota użytkowania. Dzięki deklaratywnej składni i automatycznemu zarządzaniu schematami DLT jest intuicyjne nawet dla początkujących użytkowników.
2. Skalowalność. Obsługuje duże wolumeny danych i może być wdrażane w środowiskach chmurowych.
3. Elastyczność formatów danych. Obsługuje JSON, Parquet oraz CSV jako formaty wejściowe i wyjściowe.
4. Brak zależności od infrastruktury DevOps. Można uruchamiać DLT bez konieczności korzystania z kontenerów czy dodatkowych backendów.

Podsumowanie

Data Load Tool (DLT) to wszechstronne narzędzie dla programistów Pythona, którzy chcą uprościć proces ładowania danych. Dzięki swojej elastyczności i prostocie jest idealnym rozwiązaniem zarówno dla małych projektów, jak i dużych implementacji korporacyjnych.