ETL bez ETL nástroje: Praktický příklad v SQL

Categories: TipyTags: Data, ETL

Když se řekne ETL (Extract, Transform, Load), většina lidí si představí specializované nástroje jako Keboola, Talend, Informatica, Pentaho, Microsoft SQL Server Integration Services nebo dbt. Občas existují ale okamžiky, kdy máte data v SQL databázi a z různých důvodů ETL nástroje nasadit nemůžete. Jak to řeším já u některých klientů?

Proč se obejít bez ETL nástroje?

Použít specializovaný nástroj k úpravě dat je příjemné z řady různých důvodu. Zjednodušuje život, zajišťuje správný běh ve správném pořadí, řeší chyby, notifikace, … Ne vždy je ale život tak krásný a použití SQL a uložených procedur pro ETL může být efektivní volbou. Třeba pokud:

Máte omezený rozpočet
Umíte SQL, ale s žádným ETL jste nepracovali (pak je ale mé doporučení jasné: naučte se to)
Prostředí je tak specifické, že vám neumožňuje nasadit specializované nástroje (DB server, k němu nic jiného a klient odmítá cokoliv dalšího spouštět/platit).

Pojďme se tedy podívat, jak to řeším já. V tomto případě v prostředí PostgreSQL.

Architektura řešení

Naše řešení sestává ze tří hlavních komponent:

tabulka ETL příkazů: Obsahuje příkazy k provedení jednotlivých ETL kroků
tabulka logů: Ukládá informace o provedení jednotlivých příkazů, včetně doby jejich trvání
uložená procedura: Hlavní logika ETL procesu, která provádí jednotlivé příkazy.

Krok 1: Vytvoření ETL tabulky

Nejprve potřebujeme tabulku, která bude obsahovat skripty, které chceme spouštět. Vzhledem k tomu, že je často potřebujeme spouštět v daném pořadí, tak kromě samotného skriptu obsahuje i další údaje v jednotlivých sloupcích:

id = id skriptu
name = název tabulky, která je skriptem vytvářena (název je pak využíván pro mazání dané tabulky v rámci procedury)
command = samostatný SQL skript
command_type = označení, do které vrstvy příkaz patří (ty jsou pak zpracovávány v pořadí L0 .. L3)

Copy to Clipboard

Krok 2: Vytvoření tabulky pro logy

Pro sledování výkonu a zaznamenávání událostí používám tabulku s logy. Do té si ukládám, kdy skript proběhl, jak dlouho trval a kdo jej spustil. Je mi následně podkladem pro optimalizaci či nějaký ten bug hunting.

Copy to Clipboard

Krok 3: Uložená procedura pro spuštění ETL procesu

Nyní přichází klíčová část – uložená procedura, která provede všechny ETL kroky. Tato procedura prochází příkazy v naší ETL tabulce a vykonává je v určeném pořadí. Celý kód najdete na konci článku, teď se podíváme na její jednotlivé části.

Rozbor kódu

Inicializace a příprava

Procedura začíná inicializací několika proměnných, které budou použity pro sledování času a logování. Proměnná current_user_name je nastavena na aktuálního uživatele, což nám umožňuje logovat, kdo proceduru spustil, loop proměnné zajišťují výpis aktuálního stavu běhu, časové proměnné pak počítají dobu běhu.

Copy to Clipboard

Mazání původních tabulek

Prvním krokem je mazání původních tabulek. To provádím sadou jednotlivých DROP příkazů, které si název tabulky berou z řídící tabulky. Informaci o smazání dávám do logu.

Copy to Clipboard

Zpracování jednotlivých vrstev dat

Procedura poté zpracovává jednotlivé vrstvy dat (L0, L1, L2, L3) postupně podle typu příkazu. Pro každou vrstvu je vypočítán celkový počet příkazů a následně jsou příkazy vykonávány v pořadí dle jejich ID.

Nejprve si tedy vytáhnu všechny skripty dané vrstvy a pak je pomocí smyčky spouštím. Pro informaci zároveň vypisuju, kde v rámci tabulek se aktuálně pohybuju. Níže je ukázka zpracování vrstvy L0, ale stejný postup je použit i pro další vrstvy.

Copy to Clipboard

Finální krok: Aktualizace dat

Na závěr procedura aktualizuje tabulku etl_refresh, která zaznamenává čas poslední aktualizace dat. Tu používám jednak pro rychlou kontrolu, kdy procedura naposledy proběhla (aniž bych se musel dívat do logu nebo do cronu. Právě cron totiž podle zadaných pravidel proceduru v daný čas spouští.

Copy to Clipboard

Další rozšíření

Zobrazená procedura opravdu jen řeší spouštění zpracování a nic jiného. U jiných klientů ji mám složitější. Příkladem může být zpracování dat z účetnictví Pohoda. To má pro každý účetní rok (a každou firmu) samostatnou databázi.

V rámci ETL tabulky (a procedury), tak řešením:

vytažení dat ze všech definovaných databází a spojení jednotlivých tabulek (včetně doplnění identifikací o jakou firmu, období a databázi se jedná)
logiku, zda chci stahovat všechny databáze, nebo mi pro konkrétní tabulku stačí jen údaje z posledního roku
rozdílné chování skriptů podle toho, zda je daný účetní rok už uzavřen nebo ne (očištění o řádky, které se týkají jiných účetních období)
rozšíření dimenzí pro následnou analytiku (tedy spojení různých číselníků, doplnění defaultních hodnot pro nevyplněné sloupce, atp.)

Jiným příkladem rozšíření může být automatické zasílání informací na e-mail/slack v okamžiku, kdy se něco rozbije a skripty nedoběhnou. V závislosti na prostředí k tomu mohu využít e-mailingové možnosti SQL agenta, lokální python kód nebo vše kontrolovat právě podle data aktualizace celého ETL až ve finálním reportovacím nástroji.

Shrnutí a celý kód

Použití SQL a uložených procedur pro ETL proces má své výhody i nevýhody. Je to tak trochu škrábání se levou rukou za pravým uchem, ale když to nejde lépe… Hlavní výhodou je nízká cena a relativně vysoká flexibilita. Na druhou stranu může být údržba a rozšíření takového řešení náročnější než použití specializovaného ETL nástroje. Nicméně, pro menší projekty nebo pro ty, kteří chtějí mít plnou kontrolu nad svými datovými toky, je to skvělá alternativa. Rozhodně lepší než psát složité procedury, které se vám budou debugovat výrazně hůře než jednotlivé SQL skripty. Ty můžete totiž vyvíjet postupně.

Copy to Clipboard

CREATE OR REPLACE PROCEDURE bisuperhero.bisuperhero_etl()
 LANGUAGE plpgsql
AS $procedure$
DECLARE
    command_record record;
    loop_count integer := 0;
    loop_total integer := 0;
    start_time TIMESTAMP;
    end_time TIMESTAMP;
    runtime_ms INTEGER;
    current_user_name TEXT := current_user;
BEGIN

RAISE NOTICE '================= Deleting original tables =================';	
    FOR command_record IN
        SELECT "name"
        FROM aliveplatform.bisuperhero.bisuperhero_etl
        WHERE command_type NOT LIKE '%Temp%'
    LOOP
        start_time := clock_timestamp();
        EXECUTE 'DROP TABLE IF EXISTS aliveplatform.bisuperhero.' || command_record."name";
        end_time := clock_timestamp();
        runtime_ms := EXTRACT(EPOCH FROM end_time - start_time) * 1000;
        -- Log the event
        EXECUTE format('INSERT INTO aliveplatform.bisuperhero.etl_logs (event_type, table_name, user_name, runtime_ms) VALUES (''DROP TABLE'', %L, %L, %s)',
                       command_record."name", current_user_name, runtime_ms);
    END LOOP;

RAISE NOTICE '================= Running: L0 Raw data =================';
    loop_count := 0;
    SELECT count(1) INTO loop_total FROM aliveplatform.bisuperhero.bisuperhero_etl WHERE command_type = 'L0 Raw data';
    FOR command_record IN
        SELECT command, "name"
        FROM aliveplatform.bisuperhero.bisuperhero_etl
        WHERE command_type = 'L0 Raw data'
        ORDER BY id ASC
    LOOP
        loop_count := loop_count + 1; -- Increment loop count
        RAISE NOTICE 'Currently executing row % / %: %', loop_count, loop_total, command_record."name"; -- Print current row
        start_time := clock_timestamp();
        EXECUTE command_record.command;
        end_time := clock_timestamp();
        runtime_ms := EXTRACT(EPOCH FROM end_time - start_time) * 1000;
        -- Log the event
        EXECUTE format('INSERT INTO aliveplatform.bisuperhero.etl_logs (event_type, table_name, user_name, runtime_ms) VALUES (''EXECUTE COMMAND'', %L, %L, %s)',
                       command_record."name", current_user_name, runtime_ms);
    END LOOP;

RAISE NOTICE '================= Running: L1 Cleansed data =================';
    loop_count := 0;
    SELECT count(1) INTO loop_total FROM aliveplatform.bisuperhero.bisuperhero_etl WHERE command_type = 'L1 Cleansed data';	
    FOR command_record IN
        SELECT command, "name"
        FROM aliveplatform.bisuperhero.bisuperhero_etl
        WHERE command_type = 'L1 Cleansed data'
        ORDER BY id ASC
    LOOP
        loop_count := loop_count + 1; -- Increment loop count
        RAISE NOTICE 'Currently executing row % / %: %', loop_count, loop_total, command_record."name"; -- Print current row
        start_time := clock_timestamp();
        EXECUTE command_record.command;
        end_time := clock_timestamp();
        runtime_ms := EXTRACT(EPOCH FROM end_time - start_time) * 1000;
        -- Log the event
        EXECUTE format('INSERT INTO aliveplatform.bisuperhero.etl_logs (event_type, table_name, user_name, runtime_ms) VALUES (''EXECUTE COMMAND'', %L, %L, %s)',
                       command_record."name", current_user_name, runtime_ms);
    END LOOP;

RAISE NOTICE '================= Running: L2 Integrated data =================';
    loop_count := 0;
    SELECT count(1) INTO loop_total FROM aliveplatform.bisuperhero.bisuperhero_etl WHERE command_type = 'L2 Integrated data';
    FOR command_record IN
        SELECT command, "name"
        FROM aliveplatform.bisuperhero.bisuperhero_etl
        WHERE command_type = 'L2 Integrated data'
        ORDER BY id ASC
    LOOP
        loop_count := loop_count + 1; -- Increment loop count
        RAISE NOTICE 'Currently executing row % / %: %', loop_count, loop_total, command_record."name"; -- Print current row
        start_time := clock_timestamp();
        EXECUTE command_record.command;
        end_time := clock_timestamp();
        runtime_ms := EXTRACT(EPOCH FROM end_time - start_time) * 1000;
        -- Log the event
        EXECUTE format('INSERT INTO aliveplatform.bisuperhero.etl_logs (event_type, table_name, user_name, runtime_ms) VALUES (''EXECUTE COMMAND'', %L, %L, %s)',
                       command_record."name", current_user_name, runtime_ms);
    END LOOP;

RAISE NOTICE '================= Running: L3 Consumable data =================';
    loop_count := 0;
    SELECT count(1) INTO loop_total FROM aliveplatform.bisuperhero.bisuperhero_etl WHERE command_type = 'L3 Consumable data';
    FOR command_record IN
        SELECT command, "name"
        FROM aliveplatform.bisuperhero.bisuperhero_etl 
        WHERE command_type = 'L3 Consumable data'
        ORDER BY id ASC
    LOOP
        loop_count := loop_count + 1; -- Increment loop count
        RAISE NOTICE 'Currently executing row % / %: %', loop_count, loop_total, command_record."name"; -- Print current row
        start_time := clock_timestamp();
        EXECUTE command_record.command;
        end_time := clock_timestamp();
        runtime_ms := EXTRACT(EPOCH FROM end_time - start_time) * 1000;
        -- Log the event
        EXECUTE format('INSERT INTO aliveplatform.bisuperhero.etl_logs (event_type, table_name, user_name, runtime_ms) VALUES (''EXECUTE COMMAND'', %L, %L, %s)',
                       command_record."name", current_user_name, runtime_ms);
    END LOOP;

start_time := clock_timestamp();
    EXECUTE 'DROP TABLE IF EXISTS aliveplatform.bisuperhero.etl_refresh; CREATE TABLE aliveplatform.bisuperhero.etl_refresh AS SELECT NOW() AS data_refreshed_at;';
    end_time := clock_timestamp();
    runtime_ms := EXTRACT(EPOCH FROM end_time - start_time) * 1000;
    -- Log the event
    EXECUTE format('INSERT INTO aliveplatform.bisuperhero.etl_logs (event_type, table_name, user_name, runtime_ms) VALUES (''CREATE TABLE'', %L, %L, %s)',
                   'etl_refresh', current_user_name, runtime_ms);

END;
$procedure$
;

ETL bez ETL nástroje: Praktický příklad v SQL

Proč se obejít bez ETL nástroje?

Architektura řešení

Krok 1: Vytvoření ETL tabulky

Krok 2: Vytvoření tabulky pro logy

Krok 3: Uložená procedura pro spuštění ETL procesu

Rozbor kódu

Inicializace a příprava

Mazání původních tabulek

Zpracování jednotlivých vrstev dat

Finální krok: Aktualizace dat

Další rozšíření

Shrnutí a celý kód

Menu

Novinky