Portál AbcLinuxu, 30. dubna 2025 23:00
V minulé části jsme se naučili vytvářet Transformace a nyní si z nich vytvoříme úlohu, která již bude představovat plnohodnotné ETL.
Úspěšné zvládnutí instalace a následné upravení systému vyžaduje:
Úloha, Job, je takovým managerem transformací. Umožňuje jejich spouštění, ovládání a obsluhování vstupních a výstupních souborů. Ze dvou a více transformací vytváří workflow (sériové nebo paralelní), do kterého můžete zařadit i podúlohy. Do workflow můžete vkládat scripty (SQL, JavaScript, Shell) pro komplexnější transformace. Dílčí transformace ovládá přes parametry, které se dědí z úlohy směrem dolu. Transformace naopak mohou nastavovat i proměnné v úloze, čímž komunikují s ostatními.
Úloha se umí spustit a ukončit a mezi tím vykonat spoustu užitečné práce, jako je:
Všechny tyto činnosti jsou velmi potřebné pro automatizované nahrávání/exportování dat, které si ukážeme na příkladu.
ETL nás/zákazníka může informovat o svém běhu a výsledku pomocí e-mailu. Vedle toho e-mail může být také vstupní brána pro data. Některá data mohou být distribuována pomocí e-mailu.
Dílčí data jsou uložena v jednotlivých souborech (1. soubor - informace o povaze pracoviště, 2. soubor - údaje o výrobě na daném pracovišti), ale my je potřebujeme zpracovat v jednom kroku (zajímají nás pouze vybraná pracoviště a na nich vyrobené výrobky)
Je připraveno pro některé DB systémy jako MySQL, MSSQL nebo MS Access. V testování je i PostgreSQL.
Na jednoduchém příkladu se postupně naučíme používat některé z výše uvedených schopností.
Náš příklad bude řešit tento problém: Z adresáře ~/pentaho/data_import/ importuje veškeré csv soubory a přesune je do adresáře ~/pentaho/data_complete/. Z názvu souboru určí datum a pobočku. Výsledná data nahraje do databáze včetně informace o datu a pobočce. Pro import souborů použijeme v minulé části vytvořenou transformaci. Pokud se někomu nepovedlo transformaci vytvořit, může si ji stáhnout zde a importovat do spoon pomocí File – Import from an XML file.
1. Spustíme spoon.sh, připojíme se k naší repository (pentaho_pdi)
2. Z nabídky File nebo pomocí stisknutí ctrl+alt+n vytvoříme novou úlohu Job 1, kterou si uložíme jako pobocky_vyroba
Tento krok nám do nově vytvářeného workflow předá seznam souborů, které bychom měli zpracovat.
Pokud předem nevíme, kolik souborů budeme importovat, musíme k seznamu přistupovat tak, že v něm je pouze jeden soubor a soubory importovat sekvenčně. Pokud můžeme importovat více souborů současně, pak si následující kroky můžeme paralelizovat. My se ale budeme věnovat sériovému zpracování. Výsledek Add filenames to result musíme nejdříve seřadit, vybrat první element a z názvu souboru vytáhnout jméno pobočky a datum a získané hodnoty uložit do proměnných, které budou použity v dalších krocích. Toto nám zajistí transformace a parametry úlohy.
Vytvoříme si novou transakci, kterou přidáme do úlohy pobocky_vyrob za Add filenames to result.
Výsledek by měl vypadat následovně
Nyní si pojďme jednotlivé komponenty popsat a nakonfigurovat
Get files from result převezme výsledek z nadřazené úlohy, v našem případě z pobocky_vyrob. Vzhledem k funkci komponenty ji není potřeba nikterak konfigurovat.
Sort rows přebírá výsledek Get files from result a chceme, aby nám jej seřadila. Get files from result, ale vrací více informací o souboru (cestu, název souboru, datum, typ aj). My si seřadíme výsledek podle celé cesty. Vzhledem k tomu, že importujeme soubory z jednoho adresáře u něhož neprocházíme podadresáře, pak se řazení podle cesty může zdát být zbytečné, ale nikdy nevíte, kdy se vám vstupní podmínky změní. Do pole Fields klikněte vedle 1 a z nabídky vyberte path. Směr nechme vzestupný (ascending) a rozlišovaní malých/velkých písmen můžete pominout. Vzhledem k tomu, že tato komponenta je univerzální a nemusí řešit pouze malé úlohy jako v našem případě, je nutné mít vždy na paměti povahu zpracovávaných dat. My seřazujeme 6 záznamů, ale pokud předchozí komponenta bude mít na výstupu několik stovek tisíc nebo milionů řádků (např. SQL dotaz) a data se nevejdou do RAM, pak může dojít ke swapování. Pro tento případ je k dispozici položka Sort directory, což je pracovní adresář pro naši úlohu. Další položky týkající se paměti (Sort size, Free memory threshold) je vhodné nastavovat experimentálně, až podle povahy seřazovaných dat. Výsledek by měl být následující:
Sample rows nám vybere n záznamů ze vstupního toku. My potřebujeme pouze první soubor, který dále budeme zpracovávat, takže do rozsahu výběru (Lines range) zadejte 1.
O souboru, který jsme vybrali, musíme nějak říct nadřazené úloze, aby s ním mohla pracovat. To zajistíme pomocí proměnných platných i v nadřazené úloze (tzv. parent job). V tuto chvíli nám na vstup přichází jediný záznam, z kterého potřebujeme získat informaci o názvu souboru (pro snadnější určení pobočky a datumu) a cestu k importovanému souboru (pro importní transformaci). Vytvoříme si tedy proměnné soubor_cesta a soubor. Do soubor_cesta načteme pole path a do soubor načteme filename. Obě proměnné musí být platné i v nadřazené úloze, takže ve Variable scope type vyberte Valid in the parent job.
Výslednou transformaci zařadíme do toku úlohy pobocky_vyrob za Add filenames to result (generuje zpracovávaný seznam souborů). Vraťte se tedy do pobocky_vyrob a z nabídky General přidejte za Add filenames to result komponentu transformace a propojte je. Dvakrát kliněte na nově přidanou transformaci a zvolte možnost Specify by name and directory pomocí ikonky transformace (4 zelené šipky směřující do středu) vyberte get file name. Tímto jsme naši novou transformaci zařadili do pro ni nadřazené úlohy, ze které ji budeme řídit.
Pokud se vám něco nepodařilo, pak si hotovou transformaci můžete stáhnout zde a importovat do Spoon pomocí File – Import from an XML file.
V proměnných soubor_cesta a soubor máme uložené informace o importovaném souboru. Z jeho názvu potřebujeme separovat informaci o pobočce a datu, z které a kdy pocházejí daná data. Tyto informace v dalším kroku použijeme pro rozlišení záznamů ve výsledné tabulce v databázi (to jsme v minulém díle neřešili). Předchozí komponenta get file name nám naplnila proměnnou soubor textem s názvem souboru ve formátu brno_20101128.csv . Pomocí JavaScriptu z názvu získáme tyto informace: název pobočky, rok, měsíc a den.
Z nabídky Scripting přesuňte na pracovní plochu komponentu JavaScript a propojte ji s get file name.
Ve vlastnostech úlohy vytvořte (pravá myška – job settings – Parameters) potřebné proměnné den, měsíc, rok, pobocka, soubor, soubor_cesta.
Nyní si vytvoříme vlastní script, který nám ze jména souboru vytáhne potřebné informace. Proměnná soubor obsahuje text, takže budeme řešit operace s textovými řetězci. Otevřete si komponentu JavaScript, přejmenujte ji na get info from filename a můžeme začít kódovat. Jednotlivé kroky jsou okomentované přímo v kódu
//pomocne promenne
var pobocka="", rok="", mesic="", den="";
var x=0;
//nacte obsah prametru soubor
dummy = parent_job.getVariable("soubor");
//ziskame pozici kde konci pobocka zakonceneho _
x = dummy.indexOf("_");
//z retezce vyjme podretezec s nazvem pobocky
pobocka = dummy.substring(0,x);
//zkrati nazev souboru o delku nazvu pobocky a _
dummy = dummy.substring(x,dummy.length());
//jako podretezce nacte jednotlive casti datumu o pevne delce
rok = dummy.substring(1, 5);
mesic = dummy.substring(5, 7);
den = dummy.substring(7, 9);
//ziskane hodnoty ulozi do parametru ulohy
parent_job.setVariable("pobocka",pobocka);
parent_job.setVariable("rok",rok);
parent_job.setVariable("mesic",mesic);
parent_job.setVariable("den",den);
result = pobocka;
Než se pustíme do vlastního nahrávání dat, pojďme si nejříve otestovat, co jsme doposud udělali. K tomu nám napomůže log Kettlu. Jako další komponentu do toku úlohy vložte z nabídky Utility objekt Write To Log. Jako ve velké většině jiných systémů je log obrovským zdrojem informací pro ladění a provoz systému. Kettle má několik úrovní logování s rozdílnou úrovní detailu. Pro náš případ nám postačí základní úroveň, tedy Basic. Do těla zprávy, která se bude vypisovat, umístíme všechny zjišťované proměnné den, měsíc, rok, pobocka, soubor, soubor_cesta, a to pomocí zápisu, který nám zaručí výpis jejich hodnot - ${proměnná}.
Obsah tedy bude
zpracovava se:
${pobocka} - ${rok} / ${mesic} / ${den}
A výsledné nastavení by mělo vypadat následovně:
Celá úloha by měla vypadat takto
Nyní můžeme naši úlohu spustit. Stiskněte F9 nebo zelenou šipku v horním toolbaru.
Tyto soubory byly zpracovávány
2010/12/22 14:32:01 - Add filenames to result - Starting job entry 2010/12/22 14:32:01 - Add filenames to result - Processing argument [/home/vratidlo/pentaho/data_import].. wildcard [.*\.csv] ? 2010/12/22 14:32:01 - Add filenames to result - Adding file [file:///home/vratidlo/pentaho/data_import/ostrava_20101128.csv] to result filenames 2010/12/22 14:32:01 - Add filenames to result - Adding file [file:///home/vratidlo/pentaho/data_import/brno_20101129.csv] to result filenames 2010/12/22 14:32:01 - Add filenames to result - Adding file [file:///home/vratidlo/pentaho/data_import/ostrava_20101129.csv] to result filenames 2010/12/22 14:32:01 - Add filenames to result - Adding file [file:///home/vratidlo/pentaho/data_import/brno_20101128.csv] to result filenames 2010/12/22 14:32:01 - Add filenames to result - Adding file [file:///home/vratidlo/pentaho/data_import/ostrava_20101130.csv] to result filenames 2010/12/22 14:32:01 - Add filenames to result - Adding file [file:///home/vratidlo/pentaho/data_import/brno_20101130.csv] to result filenames
Informace z našeho logu
2010/12/22 14:32:01 - Info o souboru: - zpracovava se: 2010/12/22 14:32:01 - Info o souboru: - brno - 2010 / 11 / 28
Takže, z 6 vstupních souborů byl zpracován pouze jeden, a to ten abecedně první - brno_20101128.csv. Z něj se správně separoval název pobočky brno a i jednotlivé části datumu rok: 2010, měsíc: 11, den: 28.
Dosavadní úlohu si můžete stáhnout zde a importovat do Spoon pomocí File – Import from an XML file.
Pro další část log nebudeme potřebovat takže jej vypneme. Pravou myškou klikněte na propojení mezi get info from filename a Write To Log a vyberte položku Disable hop. Tím jsme deaktivovali spouštění této komponenty. Samozřejmě, že můžete komponentu Write To Log smazat, ale může se vám ještě hodit 
Nyní víme, jak se jmenuje soubor, který chceme importovat, umíme z jeho názvu přečíst jméno pobočky a datum, pro který jsou data platná. Zbývá nám tedy jen obsah daného soubor importovat do databáze a připojit k němu informaci o pobočce a datumu. V databázi ovšem nemáme žádnou tabulku, která by byla schopna pojmout veškeré výše zmíněné informace. Vytvoříme si tedy novou a tu (pobocky_data), do které jsem v minulém díle použijeme jako dočasnou pro import nových dat. Po každém importu vymažeme její obsah.
CREATE TABLE vyroba_data ( pobocka character varying(12) NOT NULL, datum_vyroby date NOT NULL, smena smallint NOT NULL, stroj character varying(12) NOT NULL, pocet_vyrobenych_kusu integer, CONSTRAINT pk_vyroba_data PRIMARY KEY (pobocka, datum_vyroby, smena, stroj) ) WITH ( OIDS=FALSE ); ALTER TABLE vyroba_data OWNER TO pentaho_user;
Stávající transformaci (import_datapobocky) musíme upravit tak, aby poslouchala nový parametr obsahující cestu k importovanému souboru – soubor_cesta. To provedeme tak, že otevřeme transformaci a budeme editovat parametry komponenty Text file input, kde změníme položku Selected file na ${soubor_cesta}
1.Abychom zajistili, že vždy budeme data nahrávat do prázdné tabulky, zaškrtněte u Table output možnost Truncate table.
Nastavíme připojení Connection na Pentaho data a napíšeme SQL příkaz, který nám vloží data z tabulky pobocky_data do vyroba_data. Jelikož do vyroba_data musíme vložit i informaci o pobočce a datumu, musíme daný SQL script rozšířit o proměnné nesoucí tyto informace. Aby jejich zápis SQL komponenta nebrala jako text, musíme zaškrtnout možnost Use variable substitution?
SQL dotaz
INSERT INTO vyroba_data(
pobocka, datum_vyroby, smena, stroj, pocet_vyrobenych_kusu)
select '${pobocka}','${rok}-${mesic}-${den}',smena, stroj, pocet_vyrobenych_kusu
from pobocky_data;
truncate table pobocky_data;
Po dosazení hodnot z proměnných se složí například tento příkaz
INSERT INTO vyroba_data(
pobocka, datum_vyroby, smena, stroj, pocet_vyrobenych_kusu)
select 'brno','2010-11-28',smena, stroj, pocet_vyrobenych_kusu
from pobocky_data;
truncate table pobocky_data;
Výsledek tedy bude
Opět můžete otestovat funcionalitu. Po testu ale nezapomeňte vymazat obsah tabulky vyrobni_data, neboť dosud nemáme vyřešen přesun již importovaných souborů a nahrávaly by se nám stále dokola. To by ovšem vyvolalo výjimku, protože ETL by se snažilo uložit do vyroba_data. Spusťte tedy:
truncate table vyroba_data;
Podle předchozích kroků jste si vytvořili ETL, které vám importuje podle abecedy první soubor definovaného adresáře do tabulky. V daném adresáři jsou ale ještě další soubory k importu. Musíme tedy vytvořit takový proces, který cyklicky bude importovat soubory, přesouvat je do jiného adresáře tak dlouho, dokud bude mít co zpracovávat. Tedy dokud zdrojový adresář nebude prázdný. Musíme tedy vyřešit přesun importovaného souboru do adresáře ~/pentaho/data_complete/, otestovat existenci dalších souborů k importu a ukončení úlohy.
Přesun souborů do adresáře ~/pentaho/data_complete/ zajistíme pomocí komponenty Move files z nabídky File management. Přidáme ji do toku za komponentu SQL. Nastavení probíhá ve dvou krocích a to, co se bude přesouvat a kam se to bude přesouvat. Na první otázku máme snadnou odpověď neboť chceme přesunout importovaný soubor, který máme uložený v proměnné soubor_cesta. Kam ho chceme přesunout víme také. Takže pojďme si komponentu nakonfigurovat.
Otevřete si komponentu Move files a v záložce General vložte do pole File/Folder source proměnnou ${soubor_cesta}
Cílový adresář ~/pentaho/data_complete/ nastavíte v Destination folder a stiskněte Add.
Komponenta umí řešit konflikty při existenci souborů, vytvářet adresáře, soubory přejmenovávat, potažmo rozšiřovat jejich název o specifické informace, jako je třeba datum
Rozhodnutí, zda-li spustit další interaci nebo úlohu, učiníme podle existence dalších souborů k importu. Toto nám zajistí komponenta na testování obsahu adresáře Check if a folder is empty. Tu najdete v nabídce Conditions. Zařaďte ji do toku za Move files. V případě, že tato komponenta odpoví souhlasně, což znamená, že testovaný adresář je prázdný (podle určité metriky), pak by měla úloha skončit úspěšně. Stejně jako pro začátek úlohy máme komponentu Star, tak i pro zakončení existují komponenty. Konkrétně se jedná o Success z nabídky General. Pokud z nějakého důvodu potřebujeme naopak úlohu přerušit, pak z Utility můžeme použít Abort job.
My ale budeme věřit ve zdárný průběh a na pracovní plochu si ještě přesuneme Success
Každá komponenta v Kettlu svému okolí sděluje, zda-li její běh byl úspěšný, či nikoliv. To dělá pomocí vazeb, které nesou příznak result true nebo false. Doposud jsme řešili pouze úspěšné vykonání předcházející činnosti a všechny naše vazby mezi jednotlivými komponentami byly na bázi „result true“. Komponenta Check if a folder is empty, kterou testujeme obsah adresáře, vrátí při prázdném adresáři True a my tím pádem budeme chtít ukončit úlohu. Obdobně jako v předchozích případech, propojte komponentu Check if a folder is empty a Success. Pokud ovšem adresář obsahuje ještě nějaký soubor csv, pak chceme, aby se nám spustila další iterace. Znovu pomocí myšky a shift propojte Check if a folder is empty s počátkem Add filenames to result. Tím máme vytvořený cyklus. Ale pozor, Add filenames to result má stále původní seznam souborů! Jeho aktualizaci zajistíme zaškrtnutím Clear result filenames
Celá úloha pak bude vypadat takto a můžete si ji stáhnout zde
Po jejím spuštění byste měli mít v tabulce vyroba_data 204 záznamů.
pentaho_data=> select count(*) from public.vyroba_data; count ------- 204 (1 row)
Obdobně jako jsme spouštěli z řádky transformaci v minulém díle, tak i úlohu lze ovládat z řádky. Skript, ktrý nám to zajistí se jmenuje kitchten.sh.
kitchen.sh INFO 22-12 19:03:57,144 - Using "/tmp/vfs_cache" as temporary files store. Options: -rep = Repository name -user = Repository username -pass = Repository password -job = The name of the job to launch -dir = The directory (dont forget the leading /) -file = The filename (Job XML) to launch -level = The logging level (Basic, Detailed, Debug, Rowlevel, Error, Nothing) -logfile = The logging file to write to -listdir = List the directories in the repository -listjobs = List the jobs in the specified directory -listrep = List the available repositories -norep = Do not log into the repository -version = show the version, revision and build date -param = Set a named parameter= . For example -param:FOO=bar -listparam = List information concerning the defined parameters in the specified job. -export = Exports all linked resources of the specified job. The argument is the name of a ZIP file.
A stejně jako jsme si minule vytvořili automatické spouštění transakce pomocí crontabu, tak analogicky můžeme naplánovat start úlohy. Jiný způsob souštění přímo z Pentaho si ukážeme v dalších dílech.
ETL neslouží pouze k importování dat, ale také k exportu. V našem příkladu jsme zpracovávali provozní data z poboček. Ta ale musel být něčím generována. Stejně tak archivace dat může být úlohou pro ETL. S blížícím se koncem roku spoustu systémů čekají roční uzávěrky, jejichž závěry budou předávány do jiných systémů (dcera -> matka apd).
Náš příklad by mohl někomu připadat, že se jedná o kanón na vrabce, ale praxe ukazuje, že bez podobných nástrojů jako Kettle by bylo nahrávání dat z rozdílných zdrojů téměř neřešitelné. ETL jsem mocné prvky každého BI řešení a častokrát na jejich realizaci stojí a padá celé řešení. Ono pokud by správně neproběhlo nahrávání dat, tak není ani co zpracovávat.
Kettle představuje velmi šikovný nástroj na vytváření a provoz ETL. Díky platformě Java lze provozovat na různých platformách. A díky úložišti metadat v databázi je celkem snadno přenositelný.
Tento článek nemá ambici detailně popsat používání Kettlu, ale spíše vám poodkrýt možnost nahrávání dat (nemusí se jednat jen o data pro Pentaho). Pro bližší studium bych vás rád odkázal na stránky Kettlu.
Příště si povíme o datových skladech
Na shledanou s Pentaho
Autor je pracovníkem společnosti OptiSolutions s.r.o.
Společnost OptiSolutions s.r.o. se zabývá poskytováním řešení v oblasti CRM a Business Intelligence založených na Open Source technologiích SugarCRM a Pentaho.
12.1.2011 09:33
Luboš Doležel (Doli) | skóre: 98
| blog: Doliho blog
| Kladensko
Díky, opraveno.
Pentaho nas tak trosku zivi Nasi zakaznici v nem maji sva firemni data a Pentaho jim pomaha v rizeni jejich firem. Mame zakazniky, kteri v nem maji desitky GB dat a pristupuje k nim desitky uzivatelu. Pokud se jedna o datamining, tak primo Pentaho nabizi Weku, coz je sam o sobe vynikajici SW na datamining. A Weka se da prave pres Kettle propojit primo s daty a vytvorit tak automatizovany proces.
Pentaho je rozsiritelne plug-iny. Stejne tak se daji psat i vlastni komponenty do Kettlu.
. Hm, mrknu na to. Vypadá to docela zajímavě. Já už střídavě pracuji právě s Talend na jednom projektu. A abych pravdu řekl, ačkoli je to velice zajímavý produkt, tak má svá úskalí v tom, na co ho je dobré nasadit a na co ne.
Bohužel v mém případě jsem narazil na řadu bugů, které jsou na druhou stranu velice rychle opravovány. Zároveň i při požadavku na novou funkcionalitu, když jsem poskytnul základní sadu kódů pro nový plugin, byly tyto nové komponenty společností po diskuzi s vývojáři opět celkem rychle zařazeny do oficiálního repositáře (samozřejmě po úpravě mého bastlu ). Chvilkama jsem měl pocit, že by to bylo lepší celé napsat v Javě (Talend je javový projekt), a myslí si to více lidí, jelikož je to při určitém typu použití velmi problematické, resp. někdy potřebuji začít tam, kde dostupné komponenty končí, takže si stejně vytvářim vlastní tzv. rutiny, v podstatě javovské metody.
No každopádně díky za info o alternativě k Talend, Pentahoo jde na věc trochu jinak, možná v některém ohledu i lépe. Zkusím si ho nainstalit a ošahat.
Good luck s projekty.
Ladislav
ISSN 1214-1267, (c) 1999-2007 Stickfish s.r.o.