Hadoop, lehký úvod do HDFS

Měl jsem teď takovou dvoudenní chřipku... což je ideální čas, dohnat na internetu to, co normálně nestíhám, tedy videa, přednášky apod. Na Hadoop mám už políčeno cca rok, ale pořád jsem se k tomu nemohl nějak dostat. Shodou okolností jsem před pár dny narazil na stránky Big Data University, kde lidé z IBM mmj. nabízejí zdarma kurz Hadoop Fundamentals I.

Kurz je dobře udělaný (videa, přepisy textu, laby), není dlouhý (dá se zvládnout za den) a dá solidní přehled o technologiích kolem Hadoopu. Plus vše se dá vyzkoušet na VMware imagi, která je k dispozici na stránkách IBM. Byť to byl kurz krátký, byl velmi výživný, takže si sám pro sebe udělám takové jakési review, abych to pořádně vstřebal.

Co je to Hadoop?

Hadoop je framework vyvíjený pod křídly Apache Software Foundation (ASF), který umožňuje distribuované zpracování velkých data setů (Big Data).  Hadoop je založen na dvou stěžejních technologiích pocházejících od Googlu - distribuovaný filesystem Google File System (GoogleFS, GFS) a algoritmus MapReduce. Jelikož je GoogleFS proprietární, obsahuje Hadoop vlastní implementaci distribuovaného filesystemu - Hadoop Distributed File System (HDFS). Stejně tak má Hadoop vlastní implementaci MapReduce enginu nazvanou Hadoop MapReduce. Obě komponenty jsou napsané v Javě.

Hadoop Distributed File System (HDFS)

HDFS je distribuovaný filesystem, určený pro komoditní hardware (takže nepotřebuje drahé high-end servery), a který je provozovaný jako abstrakce nad nativním filesystemem. Pro ukládání souborů používá bloky o velikosti 64 MB (default), nebo 128 MB (recommended), které jsou replikovány na jednotlivé uzly Hadoop clusteru.

Replikace HDFS bloků

Jednotlivé uzly Hadoop clusteru jsou sdruženy do tzv. racků (racks), jejichž souhrn pak tvoří právě onen cluster. Rack je sdružení uzlů do logické jednotky. Pokud je potřeba síťová komunikace mezi jednotlivými uzly, je preferovaná komunikace v rámci jednoho racku. Naopak u replikace je vhodné, aby byla vytvořena do jiného racku (přesněji, jedna replika bloku je vytvořena v tomtéž racku a jedna je vytvořena v jiném).

Hadoop cluster

HDFS je založený na architektuře master/slave. Hadoop cluster obsahuje jediný master server - NameNode, který spravuje jmenný prostor a metadata filesystemu a také reguluje přístup klientů k souborům. Ostatní uzly v clusteru jsou typu DataNode a slouží k ukládání bloků dat, které pak vystavují klientům. DataNody periodicky reportují NameNodu seznam bloků, které jsou na nich uloženy.

HDFS uzly v Hadoop clusteru

HDFS commands

Pro práci s HDFS se používá podmnožina POSIX-like příkazů, jejich kompletní přehled (a reference) je na stránce Hadoop Shell Commands. Podobnou informaci můžeme dostat přímo z příkazové řádky některým z příkazů:

• hadoop fs
• hadoop fs -help
• hadoop fs -help <příkaz>

Nápověda Hadoop shellu

Obecný HDFS příkaz má tvar (důležitá je ta pomlčka před příkazem):

• hadoop fs -<příkaz>

Příklad použití HDFS příkazů je na následujícím obrázku. K tomu je potřeba poznamenat, že Hadoop přebírá uživatele a práva z hostujícího (unix) systému. Uživatel je tedy stejný jako výsledek příkazu whoami a skupiny jsou získány příkazem bash -c groups.

Příklad HDFS příkazů

V předšlém příkladu je nejprve uživatelem root vytvořen adresář /user/guido, následně je změněn vlastník adresáře na guido. Pak si již uživatel guido nakopíruje z lokálního filesystemu na HDFS soubor a zjistí jeho velikost.

Příště

V průběhu psaní se mi článek trochu rozrostl, takže jsem se rozhodl ho rozdělit do více částí. V té následující bych se chtěl věnovat Hadoop MapReduce a v ještě další (a zatím předpokládám, že závěrečné) bych chtěl probrat způsoby dotazování nad Hadoopem, tedy nástroje jako Pig, Hive a Jaql.

Maven, buildovací a konfigurační profily

Shodou okolností jsem teď byl v jednom týdnu vypomáhat na dvou projektech jako jako "problem solver", kde jsem mmj. řešil věci kolem Mavenu. V obou případech bylo potřeba vytvořit buildovací profily pro konfiguraci na různá prostředí. Chtěl bych si tady na jednom místě shrnout, jak nastavit Maven pro konfiguraci resources a webResources.

Správa konfigurací se dá v Mavenu řešit různými způsoby (řekl bych, že dost často se to dělá přes maven-antrun-plugin). Já jsem aktuálně použil filtering. Zadání je jednoduché - mám konfigurační properties soubor a potřebuju v něm mít jiné hodnoty pro různá prostředí, např. DEV a TEST.

Filtrování resources
Filtrování v Mavenu funguje následovně. Do standardního layoutu přidám adresář src/main/filters a v něm jsou filtrovací properties soubory s vkládanými hodnotami (je potřeba použít "plain" properties - s XML properties to v Mavenu nefunguje). V adresáři src/main/resources pak mám "normální" properties soubory s placeholdery pro vložení hodnot. Placeholder má formát ${property.klíč}.

Layout projektu

Obsah souboru default.properties:

my.filter.value=world

Obsah souboru hello.txt:

Hello, ${my.filter.value}!

Aby došlo k filtrování, je potřeba je mít v pom.xml zapnuté a mít uvedenou cestu k filtrovacímu souboru:

<build>
  <filters>
    <filter>
      src/main/filters/default.properties
    </filter>
  </filters>
  <resources>
    <resource>
      <directory>src/main/resources</directory>
      <filtering>true</filtering>
    </resource>
  </resources>
</build>

Pokud teď pustíme zpracování resources příkazem:

• mvn resources:resources

najdeme v adersáři target/classes soubor hello.txt s nově vloženou hodnotou:

Hello, world!

Buildovací profily
V předešlé sekci jsme do properties souboru vložili definovanou hodnotu (můžeme ji nazývat defaultní). Pro vložení jiné hodnoty je potřeba vytvořit profil v souboru pom.xml:

<profiles>
  <profile>
    <id>profile-1</id>
    <build>
      <filters>
        <filter>
          src/main/filters/profile-1.properties
        </filter>
      </filters>
    </build>
  </profile>
</profiles>

Profil zapneme přepínačem -P <profil>:

• mvn resources:resources -P profile-1

Za předpokladu, že soubor profile-1.properties vypadá takto:

my.filter.value=profile-1 world

najdeme v souboru target/classes/hello.txt nově vloženou hodnotu:

Hello, profile-1 world!

Filtrování webResources
Výše uvedený postup funguje pouze nad adresářem src/main/resources. Pokud chceme totéž provést v adresáři s web resources (WEB-INF atd.), musíme nakonfigurovat obdobným způsobem maven-war-plugin:

<build>
  <plugins>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-war-plugin</artifactId>
      <version>2.2</version>
      <configuration>
        <webResources>
          <resource>
            <directory>
              src/main/webapp/WEB-INF
            </directory>
            <targetPath>WEB-INF</targetPath>
            <filtering>true</filtering>
            <includes>
              <include>web.xml</include>
            </includes>
          </resource>
        </webResources>
      </configuration>
    </plugin>
  </plugins>
  <filters>
    <filter>
      src/main/filters/default.properties
    </filter>
  </filters>
</build>

Pro úplnost ještě doplním layout projektu:

Layout web projektu

Cesta samuraje, rok první

[update 29. 6. 2014] Změna názvu, původní titul: Změna domény a malá rekapitulace [/update]

Po roce příležitostného psaní jsem se rozhodl vzít to trochu vážně a tak jsem si pořídil pro svůj blog doménu: sw-samuraj.cz. Snad se mi podaří i další krok - psát trochu pravidelněji.

Cca po roce ještě stále hledám zaměření blogu - nechtěl jsem zakládat další blog o Javě, což je (kromě architektury) hlavní náplní mé práce. Takových (a kvalitních) blogů už je v Česku dost. Co mi naopak chybí je nějaký blog o softwarovém inženýrství, architektuře apod.

Když si zpětně zrekapituluju témata, o nichž jsem psal, tak nejvíc zápisů bylo inspirovaných odbornými knížkami, které jsem četl. Většinou byly zaměřeny na, pro mne aktuální, projektová témata. Je to dáno tím, že jsem neúnavný čtenář a čtení je pro mne jedním z primárních zdrojů informací, poznání a osobního rozvoje. Poslední rok mi v tomhle hodně pomohl Kindle.

S osobním rozvojem také souvisí další téma - certifikace. Téma pro někoho kontroverzní, pro mne užitečné. Kromě toho, že certifikace s sebou nesou jisté příjemné bonusy, jsou pro mne přínosné zejména ze studijního/znalostního hlediska, vždycky mě posunuly v dané doméně o slušný kousek dál.

Trochu mne zarazilo, že jsem málo psal o technologických a projektových záležitostech, tak to bych chtěl do budoucna napravit a více se této oblasti věnovat. Technologickým tématem číslo jedna pro mne v uplynulém roce byla určitě integrace a zejména messaging, což bude v následujícím období nejspíše pokračovat (i když půjde spíš o web services).

Na závěr bych zmínil ještě jedno téma, kterému bych se chtěl věnovat - sice jsem o něm psal jenom jednou, ale je to nejčastější zdroj návštěvníků z vyhledávačů na tomto blogu - odhady pracnosti sotfwaru. Řekl bych, že je to téma, které řadu lidí zajímá a přitom neví, jak k němu přistoupit, jak ho uchopit, kde začít apod. Takže myslím, že by mohlo být přínosné, se tomu trochu pověnovat.

Tak. Pokud letos nebude konec světa, za rok zase uvidím, kam jsem se já (i svět) posunul.

Flex, pár drobností

Při studiu na nedávnou Flexovou certifikaci jsem se musel poměrně hodně hluboko ponořit do Flexové dokumantace. A to jak do reference ActionScriptu, tak do konceptů a architektury samotného Flexu Using Flex. Důvod byl jednoduchý - nedostatek praxe jsem musel nahradit porozuměním.

Protože se mi architektura Flexu líbíla, rád bych se k němu vrátil a pro ten případ si tady chci zaarchivoval pár věcí, které mne zaujaly.

Bidirectional binding
Pokud chci mít obousměrné datové propojení dvou komponent, mám k dispozici tři způsoby. Buď je vzájemně provázat jednosměrným bindingem:

<s:TextInput id="input1" text="{input2.text}"/>
<s:TextInput id="input2" text="{input1.text}"/>

Nebo na zdroji říct, že propojení je obousměrné pomocí znaku zavináče (@):

<s:TextInput id="input1" text="@{input2.text}"/>
<s:TextInput id="input2"/>

Anebo použít tag <fx:Binding> s nastavenou twoWay property:

<fx:Binding source="input1.text"
            destination="input2.text"
            twoWay="true"/>

Event flow
Flex je framework postavený mmj. na Even-driven architekturě. Události zde proto hrají jednu z ústředních rolí. Pokud je na nějakém komponentu spuštěna (triggered) událost, má Flex definovány tři fáze, během kterých zjišťuje, zda události naslouchají nějaký event  listeneři:

Tři fáze event flow

1. Capturing Phase probíhá od kořenového uzlu (root node) komponentového stromu až k rodičovskému uzlu cílového komponentu (target node). Na obrázku Application -> Group -> Panel. Flash Player na každém uzlu kontroluje, zda má na spuštěnou událost zaregistrovaný listener, pokud ano, listener zavolá.
2. Targeting Phase zahrnuje pouze cílový/spouštěcí komponent (target node). Na komponentu jsou zavolány registrovaný listenery.
3. Bubbling Phase probíhá od rodičovského uzlu cílového komponentu zpět ke kořenovému uzlu. Na obrázku Panel -> Group -> Application. Opět jsou zavolány registrované listenery.

Jak to všechno funguje ukazuje následující kód. V MXML je definován komponentový strom z obrázku (Application -> Group -> Panel -> Button). Ve fázi inicializace (funkce initializeHandler) jsou na komponenty navěšeny listenery, na každý dva - jeden pro fázi capturing, jeden pro fázi bubbling. Zbytek funkcí jsou event handlery, které vypisují text do debug konzole (metoda trace).

<?xml version="1.0" encoding="utf-8"?>
<s:Application xmlns:fx="http://ns.adobe.com/mxml/2009" 
           xmlns:s="library://ns.adobe.com/flex/spark"
           initialize="initializeHandler(event)">

  <fx:Script>
    <![CDATA[
      import mx.events.FlexEvent;

      protected function clickHandler(
                      event:MouseEvent):void {
        trace("Button has been clicked.");
      }

      protected function groupEventHandler(
                      event:Event):void {
        trace("Event has been dispatched to: "
                        + event.currentTarget);
      }

      protected function panelEventHandler(
                      event:Event):void {
        trace("Event has been dispatched to: 
                        + event.currentTarget);
      }

      protected function buttonEventHandler(
                      event:Event):void {
        trace("Event has been dispatched to: 
                        + event.currentTarget);
      }

      protected function initializeHandler(
                      event:FlexEvent):void {
        group.addEventListener(MouseEvent.CLICK,
                               groupEventHandler,
                               true);
        group.addEventListener(MouseEvent.CLICK,
                               groupEventHandler);
        panel.addEventListener(MouseEvent.CLICK,
                               panelEventHandler,
                               true);
        panel.addEventListener(MouseEvent.CLICK,
                               panelEventHandler);
        button.addEventListener(MouseEvent.CLICK,
                                buttonEventHandler,
                                true);
        button.addEventListener(MouseEvent.CLICK,
                                buttonEventHandler);
      }

    ]]>
  </fx:Script>

  <s:Group id="group">
    <s:Panel id="panel" title="Panel">
      <s:Button id="button"
                label="Button Event"
                click="clickHandler(event)"/>
    </s:Panel>
  </s:Group>

</s:Application>

Po kliknutí na tlačítko Button Event se na debug konzoli vypíše následující výpis (řádky jsou zalomeny). Jednotlivé barvy představují výše popsané fáze.

Event has been dispatched to: Work.ApplicationSkin2
    ._ApplicationSkin_Group1.contentGroup.group
Event has been dispatched to: Work.ApplicationSkin2
    ._ApplicationSkin_Group1.contentGroup.group.panel
Button has been clicked.
Event has been dispatched to: Work.ApplicationSkin2
    ._ApplicationSkin_Group1.contentGroup.group.panel
    .PanelSkin7._PanelSkin_Group1.contents
    .contentGroup.button
Event has been dispatched to: Work.ApplicationSkin2
    ._ApplicationSkin_Group1.contentGroup.group.panel
Event has been dispatched to: Work.ApplicationSkin2
    ._ApplicationSkin_Group1.contentGroup.group

HTTPService
Pro zaslání HTTP requestu a příjem odpovědi na/z nějaké URL obsahuje Flex třídu HTTPService. Třída funfuje asynchronně - metodou send() se odešle request a události result se nastaví handler, který zpracuje response. Celé je to jednoduché a elegantní:

<fx:Script>
  <![CDATA[
    import mx.rpc.events.ResultEvent;

    protected function init():void {
      categoryService.send();
    }

    protected function resultHandler(
                            event:ResultEvent):void {
      trace(event.result);
    }
  ]]>
</fx:Script>
<fx:Declarations>
  <s:HTTPService id="categoryService"
         url="http://www.flexgrocer.com/category.xml"
         resultFormat="e4x"
         result="resultHandler(event)"/>
</fx:Declarations>

Oracle, vyhodnocení časových podmínek

Dneska jsem řešil (pro mne) zajímavý případ - potřeboval jsem v Oracle databázi postavit dotaz s následujícími parametry:

• zkontrolovat, že dva určité záznamy mají hodnotu DATE mezi 17:00 včerejšího dne a 7:00 dnešního dne,
• jako výstup dotazu mít hodnotu true nebo false (v Oraclu 1 a 0).

Dejme tomu, že máme následující tabulku

CREATE TABLE result_log (
    code VARCHAR2(20),
    logged DATE
  );

do které vložíme  dva záznamy odpovídající našemu intervalu (17:00-7:00):

INSERT INTO result_log VALUES
  ('codeA', TRUNC(sysdate - 1) + 22 / 24);
INSERT INTO result_log VALUES
  ('codeB', TRUNC(sysdate) + 5 / 24);

Výsledek by měl vypadat nějak takto:

SELECT code, TO_CHAR(logged,
                     'YYYY-MM-DD HH24:MM:SS')
             AS logged
  FROM result_log;

CODE                 LOGGED
-------------------- -------------------
codeA                2012-04-05 22:04:00
codeB                2012-04-06 05:04:00

Požadované záznamy (codeA, codeB) vybereme následujícím dotazem a zároveň je ohodnotíme, že jsou/nejsou z daného časového intervalu (1 = jsou, 0 = nejsou):

SELECT
  CASE
    WHEN (logged BETWEEN
        (TRUNC(sysdate - 1) + 17 / 24) AND
        (TRUNC(sysdate) + 7 / 24))
    THEN 1
    ELSE 0
  END AS eveluation
FROM result_log
WHERE code IN ('codeA', 'codeB');

EVELUATION
----------
         1
         1

Jelikož nad výsledky potřebujeme v podstatě provést logickou konjunkci (AND) můžeme k tomuto účelu použít funkci DECODE. Výsledný dotaz pak vypadá následovně:

SELECT DECODE(SUM(eveluation), COUNT(*), 1, 0)
  AS evaluation
FROM
  (SELECT
    CASE
      WHEN (logged BETWEEN
        (TRUNC(sysdate - 1) + 17 / 24) AND
        (TRUNC(sysdate) + 7 / 24))
      THEN 1
      ELSE 0
    END AS eveluation
  FROM result_log
  WHERE code IN ('codeA', 'codeB')
  );

EVALUATION
----------
         1

Elegantní na tomto řešení je, že stačí pouze přidat do klauzule WHERE další kód (codeC) a vše ostatní bez problémů dál funguje.

ActiveMQ, messaging podle Apache

Poslední dobou jsem se zabýval integračním projektem, který používal komerční technologii od IBM - WebSphere Messaage Broker, který jako infrastrukturu využívá WebSphere MQ. Abych měl, jako solution architekt, k tomuto řešení nějakou alternativu, rozhodl jsem se nastudovat open source messagingovou platformu ActiveMQ od Apache Software Foundation (ASF).

ActiveMQ je messagingový server postavený nad specifikací Java Message Service (JMS), a který ve spojení i frameworkem Apache Camel implementuje Enterprise Integration Patterns (EIP). Obecné vlastnosti ActiveMQ se dají shrnout do následujících bodů:

• implementace JMS 1.1,
• integrace se Springem,
• integrace s (Java) aplikačními servery,
• vlastní protokol OpenWire pro high perfomance klienty v Javě, C, C++ a C#,
• embedded broker,
• broker clustering,
• REST API,
• AJAX API,
• ad.

Výborným zdrojem informací je kniha AcitveMQ in Action přímo od autorů a přispěvatelů ActiveMQ. Vypíchnul bych z ní pár momentů, které mne zaujaly.

Komunikační protokoly
Klienti se můžou k brokeru připojit pomocí různých protokolů, které jsou vystaveny pomocí tzv. transportních konektorů. Mmj. jsou k dispozici obligátní:

• TCP
• SSL
• HTTP(S)
• UDP

Pokud klient i broker běží v jednom JVM, může klient použít VM protokol, který spustí embedded broker. Komunikace pak neprobíhá po síti, jako u ostatních typů konektorů, ale klient volá přímo metody na objektu (instanci) brokeru.

Pro konfiguraci konektoru se používá následující URI:

• scheme://host:port?queryKey=queryValue

Konfigurace několika konektorů pak může vypadat třeba takto:

<transportConnectors>
    <transportConnector
        name="openwire"
        uri="tcp://localhost:61616?trace=true"/>
    <transportConnector
        name="ssl"
        uri="ssl://localhost:61617"/>
    <transportConnector
        name="vm"
        uri="vm://localhost"/>
</transportConnectors>

Message Storage
JMS specifikace definuje dva způsoby doručení zpráv (DeliveryMode) - perzistentní a neperzistentní. Pokud jsou zpráva, nebo producent nastavený jako perzistentní, musí JMS provider zajistit bezpečné uložení zpráv (aby např. přežily výpadek serveru). ActiveMQ nabízí pro uskladnění zpráv čtyři strategie:

• KahaDB message store. Rychlá a škálovatelná file-based perzistence s transakčním žurnálem a rychlým zotavením.
• AMQ message store. Předchůdce KahaDB, obdobné vlastnosti.
• JDBC message store. Perzistence zpráv do relační databáze.
• Memory message store. Všechny pezistentní zprávy jsou drženy v paměti, tj. nejsou perzistovány ve smyslu JMS specifikace.

KahaDB se skládá ze tří komponent:

• Cache drží zprávy pro aktivní konzumenty.
• Data logy slouží jako transakční žurnál. Obsahují uložené zprávy a transakční příkazy.
• BTree indexy referencují zprávy v data logu.

Schéma KahaDB  (zdroj ActiveMQ in Action)

REST API
ActiveMQ poskytuje jednoduché API pro publikaci a konzumaci zpráv RESTovým způsobem. Vzhledem k tomu, že JMS poskytuje pouze dvě operace - send a receive - je mapování na HTTP velmi přímočaré. Pro publikování zpráv je to (HTTP) POST, pro jejich konzumaci (HTTP) GET nebo DELETE.

Mapování na URI pak vypadá takto:

• http://host:port/queue
• http://host:port/topic/subtopic

High Availability
Pokud budeme chtít zajistit vysokou dostupnost našeho messagingového řešení, budeme potřebovat několik brokerů běžících na různých strojích. Něco jako:

High Availability v režii ActiveMQ je zajištěna dvěma typy Master/Slave konfigurací:

• Shared nothing
• Shared storage

U shared nothing konfigurece má master i slave vlastní message store. Slave se po startu připojí k masteru a veškeré stavy masteru (zprávy, akcknowledgements, transakce atd.) jsou replikovány na slave. Když master spadne, jsou všichni klienti přesměrováni (pomocí fail over protokolu) na slave, který se stává novým masterem. Konfigurace fail over protokolu vypadá takto:

• failover://(tcp://master:61616,tcp://slave:61616)

Omezením tohoto řešení je, že master může mít definovaný pouze jeden slave (a slave nemůže mít definovaný další vlastní slave).

Shared nothing master/slave (zdroj ActiveMQ in Action)

Shared storage konfigurace umožňuje, aby bylo definováno více slave brokerů, které spolu s masterem, sdílejí společný storage mechanismus - tím může být sdílený filesystem, nebo relační databáze. Master má pak storage zamknutý. Když master spadne, jeden ze slave brokerů se stane novým masterem a zamkne si storage pro sebe.

Shared storage master/slave (zdroj ActiveMQ in Action)

Závěrem
Pokud to mám říct jednou větou - ActiveMQ se mi architektonicky/technologicky líbí a pokud bych na nějakém projektu potřeboval řešit messaging v Javě, určitě by to byl žhavý kandidát. Taky mi to nedá, abych nesrovnal ActiveMQ s WebSphere MQ. Co se týče funkčností, jsou obě řešení srovnatelná. Veliký rozdíl ovšem bude v pracnosti a nákladech - ActiveMQ půjde implementovat nesrovnatelně levněji a rychleji. No a samozřejmě cena, zde je rozdíl astronomický - ActiveMQ je zdarma, WebSphere MQ bude stát řádově miliony korun jenom na licencích.

Další cesta, jak rozvíjet znalosti o ActiveMQ je celkem jednoznačná - Apache Camel, což je implementace EIP od ASF, která řeší věci jako vytváření zpráv, jejich směrování, transformaci, orchestraci ad.