Konténerezett Hadoop és Cassandra cluster konfigurálása - harmadik rész

A sorozat előző részeiben (1, 2) Vagrantos környezetben felépítettünk egy Hadoop clustert. Ebben a befejező cikkben egy Cassandra fürtöt fogunk telepíteni, majd egy map/reduce jobot futtatunk a teljes clusteren. Izgalmasan hangzik, vágjunk is bele.

Előkészítés

Első dolgunk, hogy meglévő projektünket frissítjük a megfelelő verzióra, és építsük újra a konténereket:

cd docker-cassandra && git checkout 2.6.0-cassandra && git pull origin 2.6.0-cassandra
cd ../hadoop-docker && git checkout 2.6.0-cassandra && git pull origin 2.6.0-cassandra
cd .. && git checkout 2.6.0-cassandra && git pull origin 2.6.0-cassandra

Élesszük fel a gépeket:

vagrant halt && vagrant up

Majd a futó Vagrantos környezetben építsük újra a konténereket:

vagrant provision master slave1 slave2 slave3

Lépjünk be a virtuális gépekre, és töröljünk minden futó konténert:

docker rm -f $(docker ps -qa)

Ha mindezzel végeztünk, hozzuk létre ismét a Weave hálózatot és a Swarm clustert, a Hadoop konténereket egyelőre hagyjuk parlagon.

Változások

Csökkentettem a master nevű gép memória igényét, mivel a továbbiakban csak mesterként fog szolgálni a benne futó Hadoop konténer, és ezzel egy időben növeltem a slaveX nevű gépek memóriáját, mert az eddigi beállítás ki volt hegyezve a Hadoopra, mostantól viszont a Cassandrának is kell helyet szorítani. A fejlesztés előrehaladtával a gépemben lévő 8 Gb RAM már sokszor kevésnek bizonyult, elkezdte a Vagrant aktívan használni a swapet, ami igen rossz hatással van a jobok futtatására, lépten nyomon elhasaltak. Én átmenetileg a 3-s számú slave gépet kikapcsoltam. Összességében 4 virtuális gépből és 15 konténerből áll a cluster, szóval személy szerint nem is csodálkozom, hogy ilyen mértékben megnövekedett a gép igény.

bootstrap.sh

if [ -n "$MASTER_IS_SLAVE_TOO" ]; then
    echo $HOST_NAME > $HADOOP_PREFIX/etc/hadoop/slaves
else
    echo "" > $HADOOP_PREFIX/etc/hadoop/slaves
fi

Bevezettem egy környezeti változót (MASTER_IS_SLAVE_TOO) melynek hatására a mester konténer szolga is lesz egyben, a változó nélkül csak mesteri teendőit látja el.

Dockerfile

RUN sed -i "s|^# Extra Java CLASSPATH.*|&\nexport HADOOP_CLASSPATH=/usr/share/cassandra/*:/usr/share/cassandra/lib/*:\$HADOOP_CLASSPATH|" $HADOOP_PREFIX/etc/hadoop/hadoop-env.sh

Javítottam a HADOOP_CLASSPATHon, hiányzott egy Cassandrás függőség.

cassandra-clusternode.sh

if [ -n "$PUBLIC_INTERFACE" ]; then
    IP=$(ifconfig $PUBLIC_INTERFACE | awk '/inet addr/{print substr($2,6)}')
    PUBLIC_IP=$IP
fi
if [ -n "$PUBLIC_IP" ]; then
    sed -i -e "s/^# broadcast_address: 1.2.3.4/broadcast_address: $PUBLIC_IP/" $CASSANDRA_CONFIG/cassandra.yaml
fi

A Cassandrás konténerben egy új környezeti változóval, név szerint PUBLIC_INTERFACE, megoldottam, hogy a Cassandra a megfelelő IP címet használja minden nemű kommunikációhoz.

if [ -n "$CASSANDRA_SEEDS" ]; then
    for a in $(echo $CASSANDRA_SEEDS | sed 's/,/ /g'); do CASSANDRA_SEEDS=$(echo $CASSANDRA_SEEDS | sed "s/$a/$(ping -c1 $a | grep PING | awk '{ print $3 }' | sed "s/(//;s/)//")/"); done
fi

Mivel a konténerek dinamikusan kapnak IP címet, a Cassandra viszont csak IP alapján tud kapcsolódni a seed szerverekhez, ezért meg kellett trükköznöm a CASSANDRA_SEEDS változót, domain neveket és IP címeket is elfogad egyaránt, majd a Cassandra indítása előtt feloldja a domain neveket IP címekre.

Futtatás

slave1

nohup docker -H tcp://192.168.50.15:1234 run --name cassandra-slave1 --dns 192.168.50.15 -h cassandra1.lo -e "PUBLIC_INTERFACE=eth0" -e "CASSANDRA_CLUSTERNAME=HadoopTest" -e "CASSANDRA_TOKEN=-9223372036854775808" -t mhmxs/cassandra-cluster > cassandra.log 2>&1 &
docker -H tcp://192.168.50.15:1234 run --name hadoop-slave1 --dns 192.168.50.15 -h slave1.lo -e "MASTER=master.lo" -e "SLAVES=slave1.lo,slave2.lo,slave3.lo" -it mhmxs/hadoop-docker:2.6.0 /etc/bootstrap.sh -bash

A Cassadnrás konténer logját a cassandra.log fájlban találjuk, érdemes a Hadoop cluster elindítása előtt összeállítani a Cassandra clustert (vagy külön terminálban nohup nélkül indítani), mert ha valami időzítési vagy hálózati probléma miatt nem találták meg egymást a nodeok, akkor elég kényelmetlen a Swarm clusterből törölgetni, majd újraindítgatni a megfelelő konténereket. Sokszor kellett a fejlesztés alatt hasonlót csinálnom, és a rendszer sem túl hiba tűrő, úgyhogy rászoktam, hogy minden lépés előtt ellenőrzöm, hogy az elemek a helyükre kerültek-e.

slave2

nohup docker -H tcp://192.168.50.15:1234 run --name cassandra-slave2 --dns 192.168.50.15 -h cassandra2.lo -e "PUBLIC_INTERFACE=eth0" -e "CASSANDRA_CLUSTERNAME=HadoopTest" -e "CASSANDRA_SEEDS=cassandra1.lo" -e "CASSANDRA_TOKEN=-3074457345618258603" -t mhmxs/cassandra-cluster > cassandra.log 2>&1 &
docker -H tcp://192.168.50.15:1234 run --name hadoop-slave2 --dns 192.168.50.15 -h slave2.lo -e "MASTER=master.lo" -e "SLAVES=slave1.lo,slave2.lo,slave3.lo" -it mhmxs/hadoop-docker:2.6.0 /etc/bootstrap.sh -bash

slave3

nohup docker -H tcp://192.168.50.15:1234 run --name cassandra-slave3 --dns 192.168.50.15 -h cassandra3.lo -e "PUBLIC_INTERFACE=eth0" -e "CASSANDRA_CLUSTERNAME=HadoopTest" -e "CASSANDRA_SEEDS=cassandra1.lo" -e "CASSANDRA_TOKEN=3074457345618258602" -t mhmxs/cassandra-cluster > cassandra.log 2>&1 &
docker -H tcp://192.168.50.15:1234 run --name hadoop-slave3 --dns 192.168.50.15 -h slave3.lo -e "MASTER=master.lo" -e "SLAVES=slave1.lo,slave2.lo,slave3.lo" -it mhmxs/hadoop-docker:2.6.0 /etc/bootstrap.sh -bash

master

docker -H tcp://192.168.50.15:1234 run --name hadoop-master --dns 192.168.50.15 -h master.lo -e "SLAVES=slave1.lo,slave2.lo,slave3.lo" -v /vagrant:/vagrant -it mhmxs/hadoop-docker:2.6.0 /etc/bootstrap.sh -bash

Job

A hozzuk létre a projekt könyvtárban a KeyCollector.java fájlt az alábbi tartalommal:

import java.io.IOException;
import java.util.*;
import java.nio.ByteBuffer;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.hadoop.*;
import org.apache.cassandra.db.*;
import org.apache.cassandra.utils.ByteBufferUtil;

public class KeyCollector {

    public static void main(String[] args) throws IOException {
        if (args.length != 1) {
            System.err.println("Usage: KeyCollector <output path>");
            System.exit(-1);
        }

        JobConf conf = new JobConf(KeyCollector.class);
        conf.setJobName("KeyCollector");

        ConfigHelper.setInputInitialAddress(conf, "cassandra1.lo");
        ConfigHelper.setInputColumnFamily(conf, "HadoopTest", "content");
        ConfigHelper.setInputPartitioner(conf, "org.apache.cassandra.dht.Murmur3Partitioner");
        SlicePredicate predicate = new SlicePredicate().setColumn_names(Arrays.asList(ByteBufferUtil.bytes("text")));
        ConfigHelper.setInputSlicePredicate(conf, predicate);

        conf.setInputFormat(ColumnFamilyInputFormat.class);

        conf.setMapperClass(KeyCollectorMapper.class);

        FileOutputFormat.setOutputPath(conf, new Path(args[0]));

        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(IntWritable.class);

        conf.setReducerClass(KeyCollectorReducer.class);

        JobClient.runJob(conf);
    }

    public static class KeyCollectorMapper extends MapReduceBase implements Mapper<ByteBuffer, Map<ByteBuffer, BufferCell>, Text, IntWritable> {
        public void map(ByteBuffer key, Map<ByteBuffer, BufferCell> columns, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
            String textKey = ByteBufferUtil.string(key);
            output.collect(new Text(textKey), new IntWritable(1));
        }
    }

    public static class KeyCollectorReducer extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
        public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text,IntWritable> output, Reporter reporter) throws IOException {
            int sum = 0;
            while (values.hasNext()) {
                sum += values.next().get();
            }
            output.collect(key, new IntWritable(sum));
        }
    }
}

Valószínűleg ez a világ legértelmetlenebb map/reduce jobja, összegyűjti a column familybe lévő kulcsokat, de ez tűnt a legegyszerűbb implementációnak. Természetesen lett volna lehetőség a job eredményét a Cassandrába tárolni, de a letisztultság jegyében, én a fájl rendszert preferáltam. Fordítsuk le az osztályt és csomagoljuk be egy jar-ba a mester Hadoop konténerben.

yum install -y java-1.8.0-openjdk-devel
cd /vagrant
mkdir build
classpath=.
for jar in /usr/share/cassandra/*.jar; do classpath=$classpath:$jar; done
for jar in /usr/share/cassandra/lib/*.jar; do classpath=$classpath:$jar; done
for jar in `find /usr/local/hadoop/share/hadoop/ *.jar`; do classpath=$classpath:$jar; done
javac -classpath $classpath -d build KeyCollector.java
jar -cvf KeyCollector.jar -C build/ .

Következő lépés, hogy ellenőrizzük a cluster működését, és teszt adattal töltjük fel az adatbázist, szintén a mester konténerből.

nodetool -h cassandra1.lo status
cassandra-cli -h casandra1.lo
create keyspace HadoopTest with strategy_options = {replication_factor:2} and placement_strategy = 'org.apache.cassandra.locator.SimpleStrategy';
use HadoopTest;
create column family content with comparator = UTF8Type and key_validation_class = UTF8Type and default_validation_class = UTF8Type and column_metadata = [ {column_name: text, validation_class:UTF8Type} ];
set content['apple']['text'] = 'apple apple red apple bumm';
set content['pear']['text'] = 'pear pear yellow pear bumm';

Elérkezett a várv várt pillanat, futtathatjuk a jobot.

$HADOOP_PREFIX/bin/hadoop jar KeyCollector.jar KeyCollector output
$HADOOP_PREFIX/bin/hdfs dfs -cat output/*

A kimenetben láthatjuk, hogy egy darab apple és egy darab pear kulcs van az adatbázisban.

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/usr/local/hadoop-2.6.0/share/hadoop/common/lib/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/usr/share/cassandra/lib/logback-classic-1.1.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
apple: 1
pear: 1

Teljesítmény

Befejezésül ejtsünk pár szót a rendszer teljesítmény optimalizálásáról.

A Cassandra elég korai verziójában bevezetésre kerültek az un. vNodeok, amik azt biztosítják, hogy egy Cassandra valós node a token tartomány több szeletét is birtokolhassa egyszerre. A Cassandra oldaláról számos előnye van ennek a megoldásnak, viszont a Hadoop felöl érkezve kifejezetten káros hatása van. A vNodeok száma kihatással van a szeletek (split) számára, ami annyit tesz, hogy annyi darab szelet egészen biztos lesz, ahány vNode van engedélyezve.

A következő paraméter, amire érdemes figyelni, az a szeletek mérete. Az alapértelmezett szelet mérete 64K, ha mondjuk van 5,000,000,000 sor az adatbázisban, akkor 64,000-el számolva 78,125 szeletben fogja elvégezni a műveletet a Hadoop, ami szeletenkénti 10 másodperccel szorozva száz óra körüli futás időt eredményez. Az alábbi sorral konfigurálhatjuk a szeletek méretét.

ConfigHelper.setInputSplitSize(job.getConfiguration(), 10000000);

Mivel a feldolgozó egységekben korlátozott mennyiségű memória áll rendelkezésre, javasolt finom-hangolni az egyszerre feldolgozott adatok számát, a példában 100-as kötegekben fogja a CQL driver a sorokat prezentálni, az alapértelmezett 1000 darab helyett.

CqlConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), Integer.toString(100));

A negyedik megkerülhetetlen téma az un. data locality. A jó teljesítmény eléréséhez elengedhetetlen, hogy az adatokat a legkisebb mértékben kelljen mozgatni. A Cassandra fejlesztői az InputFormat implementálása során szerencsére gondoltak erre a probléma forrásra, annyi a dolgunk, hogy ugyanarra a hostra telepítjük a Cassandrát, és a Hadoop JobTrackert. Sajnos Docker konténereket használva ezt bukjuk, mert a két konténer mindig két különálló host lesz, gyakorlatilag mintha külön gépen futnának a szolgáltatások. Megtehetnénk, hogy összeházasítjuk a két konténert, de akkor a Docker szemlélettel mennénk szembe, miszerint minden szolgáltatás egy önálló konténer legyen. Ha ezt el szeretnénk kerülni, akkor bizony be kell koszolnunk a kezünket. Én első lépésben megnyitottam a legfrissebb dokumentációt. Láthatjuk, hogy semmi nem található benne a témával kapcsolatban, de ne keseredjünk el, ugyanis van egy őse ennek a dokumentumnak, nyissuk meg a AbstractColumnFamilyInputFormat fájlt is. A public List<InputSplit> getSplits(JobContext context) metódusban történik a csoda, itt állítja össze a Hadoop a szeleteket, minden egyes szeletet egy InputSplit objektum reprezentál, és tartalmazza azt/azokat a host neveket, ahol az adat megtalálható.

A jelen cluster egy szoftveres hálózaton kommunikál, így ebben az esetben nincs jelentősége az adatok tényleges helyének, nem is bonyolítanám tovább a rendszert saját InputFormat írásával és/vagy különböző DNS trükkök bevezetésével. A jó megoldást mindenféleképpen az adott hálózati architektúrának megfelelően kell kialakítani. Akit érdekelnek további a teljesítménnyel kapcsolatos kérdések itt talál pár választ.

JVM futásidejű monitorozása

Figyelem az alábbi bejegyzés nyomokban fizetetlen reklámot és szubjektív véleményt tartalmaz.

Egy ideje a JVM valós idejű monitorozásának lehetőségeit kutattam, és csak hosszas keresés után találtam meg a projektnek, és a költség tervnek megfelelőt.
Szerencsére megoldás akad bőven a fizetőstől az ingyenesen át a szabadig megoldásokig. A teljesség igénye nélkül szeretnék bemutatni néhányat közülük.

• Dynatrace kétségtelen, hogy egyike a legprofibb megoldásoknak. Volt szerencsém egy hosszabb lélegzet vételű prezentációt végigülni, ahol ebből a csodás eszközből kaptunk ízelítőt. Gyakorlatilag a monitorozás, hiba feltárás, és reprodukálás mekkája a Dynatrace. Központilag telepített szerver gyűjti az információkat a különböző ügynököktől (Java, .NET, böngésző), majd ezeket az információkat un. PurePathokba szervezi, ahol nyomon tudjuk követni egy kérés teljes útvonalát az infrastruktúrában. Pontos képet kaphatunk, hogy mely rétegben mennyi időt töltött a kiszolgálás, és a PurePathon belül minden irányban teljes átjárást biztosít a rendszer, ami azt jelenti, hogy pár kattintással el lehet érni a végrehajtó kód sort, a futtatott lekérdezéseket, a felhasználót, és annak a többi kérését, és még sorolhatnám. Profizmusához mérten van az árazása, természetesen létezik ingyenes próba időszak, cserébe viszont képzett kollégák segítenek a rendszert beüzemelni.
  

• Következő alternatíva a New Relic. Mi aktívan használjuk több projekten is, és alapvetően meg vagyunk vele elégedve. Ára sokkal pénztárca barátabb, mint a Dynatracé, de tudása is ezzel arányosan kevesebb. A New Relic képes monitorozni a szervert (mi Linuxot használunk, nincs tapasztalatom egyéb operációs rendszerekkel), az adatbázist, a JVMet, kéréseket, hibákat, majd ezekből tetszetős grafikonokat rajzol. Létezik hozzá mobil alkalmazás és böngésző monitorozó eszköz is. Hátránya, hogy a New Relic szerverei felé jelentenek az ügynökök, így egyfelől van egy minimális késleltetése, másfelől a weboldaluk sebessége is hagy némi kívánnivalót maga után. Kevésbé ár érzékeny projektek esetén kiváló választás lehet.
• Az AppDynamicsról olvastam még eléggé pozitív cikkeket, sajnos saját tapasztalatom nincs velük kapcsolatban.
• Utolsó fizetős megoldás az előzőekhez képest még szerényebb a nyújtott szolgáltatások terén, de olyan kedvező a fizetési modelljük, hogy mindenféleképpen érdemes őket megemlíteni. A SemaText  szinte valósidejű monitorozást végez fáék egyszerűséggel, de a támogatott platformok igen széles palettán mozognak: AWS, Apache, Cassandra, Elasticsearch, HAProxy, HBase, Hadoop-MRv1, Hadoop-YARN, JVM, Kafka, Memcached, MySQL, Nginx, Redis, Sensei, Solr, SolrCloud, Spark, Storm, ZooKeeper. Az ingyenes verzióban 30 percig vissza menőleg örzik meg az adatokat, és még van pár limitáció, de alkalmazásonként választhatunk csomagot, és akármikor felfüggeszthetjük egy alkalmazás/cluster monitorozását (órában van megadva a legkisebb fizetési egység). Amit mindenféleképpen, mint előnyt meg szeretnék említeni, hogy APIjukon keresztül saját metrikékat is viszonylag kényelmesen megjeleníthetünk. Hátránya, hogy a New Relichez hasonlóan az ő szervereik tárolják az adatokat, és az ügynöknek, amit telepíteni kell, rengeteg a csomag-függősége, legalábbis Linuxon.

A nyílt forrású megoldások esetén a tanulási görbével, és a beüzemelés költségével fizetjük meg az árát a monitorozásnak (igaz ezt csak 1x kell). Azt tapasztaltam, hogy képességeikben elmaradnak fizetős társaiktól, de ami sokkal nagyobb probléma (szerintem), hogy nincs próbaidő. Nem tudom egy teszt szerveren kipróbálni, nincs hozzájuk demó felület, amit meg lehetne nyomkodni.

• Első delikvens a JavaMelody, aminek a telepítése igen egyszerű, a letöltött war fájlt deployoljuk az alkalmazás-szerveren. Hátránya, hogy csak lokális monitorozást végez, ami több mint a semmi, de csak egy hajszállal.
• A  stagemonitor igen ígéretes projektnek tűnik, kár, hogy csak a JVM helyi megfigyelésére alkalmas, és számunkra csak központosított megoldások jöhetnek számításba. A weboldalt böngészve láthatjuk, hogy igen széles spektrumon követi nyomon az alkalmazás működését, és gyönyörű grafikonokon ábrázolja az adatokat.
• Legtöbben az interneten a JAMont ajánlották, ami egy Java monitorozásra alkalmas API. A dokumentációból első olvasásra kiderült számomra, hogy telepítése nem triviális, és a metrikák pontos megtervezése után az alkalmazásban implementálni is kell azokat. Őszinte leszek nem ugrottam fejest a JAMon világába. Biztos nagyon szép és nagyon jó, de a csapat "produktivitását" nem növeli, az ügyfélnek nem eladható hogy x hétig metrikákat reszelgetünk, meg grafikonokat rajzolgatunk.
• Nyílt forrású megoldások közül nekem a MoSKito tűnik a legkiemelkedőbbnek, sajnálom, hogy későn akadtam rá, és addigra már belaktunk egyéb szolgáltatásokat. A MoSKitó kifejezetten Java fürtök valós idejű megfigyelését célozza meg.

Talán ebből a bejegyzésből, és kiderül (legalábbis remélem), hogy nincs szent grál a témában, mert az egyik drága, főleg ha automatikusan skálázódó alkalmazást szeretnénk monitorozni, ahol percek alatt 5-8-ról 30-50-re nőhet a JVMek száma, van amelyik csak lokálisan működik, míg másnak a beüzemelése visz el túlzottan sok erő forrást. Egyesek képesek kontextusban látni az alkalmazást, míg mások csak számokat vizualizálnak. Mindenféleképpen érdemes alaposan körbejárni a témát, és az igényeknek leginkább megfelelőt választani.

Groovy funkcionális eszközök

A Groovy sorozatot folytatva (1, 2) ebben a bejegyzésben a nyelv funkcionális aspektusát szeretném bemutatni a teljesség igénye nélkül. Két fontos tulajdonság képezi az alapját a funkcionális programozásnak Groovyban, az egyik, hogy van lehetőség anonim funkciók (Closure) írására, a másik pedig, hogy az utolsó kifejezés értéke automatikusan visszatérési érték lesz, ha nincs explicit visszatérési érték meghatározva (üres Closure visszatérési értéke null). Ez a két tulajdonság elengedhetetlen ahhoz, hogy funkcionális szemléletben tudjunk programozni, de pusztán e két dolog használata még nem eredményezi automatikusan a szemlélet-váltást. Vegyük sorra milyen egyéb eszközökkel támogatja a Groovy a munkánkat.

• Először nézzük a Closure összefűzést:

  def m = { "${it}-" }
  def p = { "${it}+" }
  def pm = m << p
  
  println pm('') == m(p('')) // true
  

  Természetesen a másik irányba is működik a dolog:

  def mp = m >> p
  println p(m('')) == mp('') // true
  

• A Closure.curry() metódus becsomagolja a Closure példányt, és az eredeti Closure paraméterei helyére lehet fix értékeket beállítani. A példa magáért beszél:

  def plus = { int f, int s, int t ->
      println "$f $s $t"
      return f + s + t
  }
  def fix = plus.curry(0, 0) // további opciók: ncurry() és rcurry()
  println fix(5) // 0 0 5
  

• Felmerülhet az igény, hogy már meglévő metódusokból csináljuk Closuret. Nem a legelegánsabb megoldás, de mindenféleképpen hasznos ha meglévő eszközeinket szeretnénk "modernizálni":

  class o {
      void f() {
          println 'called'
      }
  }
  
  def c = o.&f // vagy new o().&f
  
  println c.class // class org.codehaus.groovy.runtime.MethodClosure
  

• A funkcionális programozásra igen jellemző a rekurzív végrehajtás, és ezen programozási nyelvek részét képezik a különböző optimalizációs eszközök. Természetesen a Groovyban is van lehetőségünk finomhangolni rekurzióinkat. Az első ilyen eszköz, amit bemutatok a Closure.memoize(), ami nemes egyszerűséggel annyit tesz, hogy a visszaadott csomagoló Closure gyorsítótárazza a végrehajtás eredményeit. Különös tekintettel kell lennünk használatakor a memória-szivárgásokra, mert alapértelmezetten nincs méret határ szabva a gyorsítótárnak:

  def a = { print "called" }.memoize() // vagy memoizeBetween(int protectedCacheSize, int maxCacheSize)
  a();a() // called
  

  def a = { i -> print "called $i " }.memoize() // vagy memoizeAtLeast(int protectedCacheSize) és memoizeAtMost(int maxCacheSize)
  a(1);a(2);a(2) // called 1 called 2
  

  Meglévő metódusainkat pedig a @Memorized annotációval tudjuk hasonló működésre bírni, mely két opcionális paramétert vár a maxCacheSize és a protectedCacheSize.
• A rekurzív hívásoknak van egy igen kártékony mellékhatása a JVMben. Minden egyes hívás rákerül a stackre, ami könnyen StackOverflowErrorhoz vezet. Ezt elkerülendő a Closure.trampoline() segítségével referenciát szerezhetünk egy olyan TrampolineClosurera, mely szekvenciálisan hívja az eredeti Closuret egészen addig, míg az TrampolineClosuret ad vissza. Ezzel a technikával mentesíti a stacket, lássuk ezt a gyakorlatban:

  def s
  s = { l, c = 0 ->
      l.size() == 0 ? c : s(l.tail(), ++c)
  }.trampoline()
  
  println s(1..10) // 10
  

  A Closure.trampoline() metódus mintájára az @TailRecursive annotációt használhatjuk, a dokumentációban szereplő megkötésekkel.
• A funkcionális nyelvek általában az un. lazy evaluation szemléletet követik, ami röviden annyit jelent, hogy csak akkor értékel ki a rendszer valamit, ha arra feltétlenül szükség van. A Groovy is biztosít megoldásokat a paradigma követéséhez.

  def l = [].withDefault { 45 }
  println l[3] // 45
  println l // [null, null, null, 45]
  

  Vagy a @Lazy annotációval tetszőleges adattagot varázsolhatunk lusta kiértékelésűre. Egy opcionális paraméterével akár puha referenciában is tárolhatjuk az értéket, természetesen az alapértelmezett működés szerint erős referenciát alkalmaz:

  class Person {
      @Lazy(soft = true) pets = ['Cat', 'Dog', 'Bird']
  }
  
  def p = new Person()
  println p.dump() // <Person@7b073071 $pets=null>
  p.pets
  println p.dump() // <Person@18e30556 $pets=java.lang.ref.SoftReference@3f0e6ac>
  

  Annyit mindenféleképpen meg kell még jegyeznem, hogy ha a mező statikus, és nem puha referenciában történik a tárolás, akkor a Initialization on demand holder mintát követi a fordító.

A beépített funkciók után térjünk át a haladó technikákra. Bár a Groovynak szoros értelemben nem része a GPars keretrendszer, mégis érdemes kicsit közelebbről megismerkedni vele. A dokumentációból idézve:

"GPars is a multi-paradigm concurrency framework, offering several mutually cooperating high-level concurrency abstractions, such as Dataflow operators, Promises, CSP, Actors, Asynchronous Functions, Agents and Parallel Collections."

• Meglévő Closurejainkat könnyedén aszinkron hívásokká alakíthatjuk a GParsExecutorsPool segítségével, ahogy a példa is mutatja.
• Collectionök párhuzamos feldolgozására a GParsPoolt tudjuk segítségül hívni. A GParsPool osztály ParallelArray alapon végzi a műveleteket, míg párja a GParsExecutorsPool hagyományos thread poolokat használja.
• A GPars része egy a Java Fork/Join könyvtárára épülő magasabb absztrakciós réteg. Ez a köztes réteg -mely a mindennapi fejlesztést hivatott megkönnyíteni- használata során nem kell szálakkal, poolokkal, és szinkronizációval bajlódnunk. Részletek a dokumentációban találhatók.
• A Dataflow egy alternatív párhuzamos feldolgozási szemlélet. Szépsége az egyszerűségében rejlik, apró párhuzamos feladatokra bonthatjuk az alkalmazásunkat, és amikor az egyik darabka még egy ki nem értékelt adathoz szeretne hozzáférni, akkor blokkolt állapotba kerül amíg egy másik ki nem értékeli azt. Működéséből adódóan nincs verseny helyzet, nincs Dead és Live Lock sem többek között. Megkötés, hogy csak egyszer adhatunk értéket a DataflowVariable életciklusa során.
• Az Agentek szál-biztos, nem blokkoló párhuzamosítást tesznek lehetővé, és ehhez annyit kell tennünk, hogy az osztályunkat a groovyx.gpars.agent.Agentből származtatjuk. Fontos különbség a Dataflow modellhez képest, hogy az Agent tartalma tetszőlegesen változtatható.
• Természetesen elmaradhatatlan kellék a méltán népszerű Actor modell. Leegyszerűsítve az Actorok üzeneteket fogadnak, továbbítanak, és válaszolnak azokra. Minden üzenet bekerül egy sorba, majd onnan a feldolgozásra. A megoldás előnye, hogy implicit szál-biztos, nem blokkolt, és nincs szükség szinkronizációra sem a soros feldolgozás miatt. Lényeges tulajdonsága az Actor rendszernek, hogy nem hagyományos szál-kezelést használ, hanem saját maga menedzseli a feladatokat. Létezik állapot-tartó, és állapot-mentes Actor egyaránt.

Ahogy a Barátkozás a Groovyval bejegyzésben is leírtam, a Groovy nem kezdő programozók kezébe való eszköz. Szép és jó ez a sok beépített okosság, de az alapok ismerete nélkül csak még jobban összezavarnak mindent, megnehezítik a hiba feltárást és az elhárítást is. Remélem sikerült kedvet csinálnom a téma mélyebb megismeréséhez, ráadásként pedig egy igazi "ínyencséget" hagytam:

def deliver(String n) {
 [from: { String f ->
     [to: { String t ->
         [via: { String v ->
             println "Delivering $n from $f to $t via $v."
         }]
     }]
 
 }]
}
deliver "Béla" from "Mezőberény" to "Kisfái" via "Traktor" // Delivering Béla from Mezőberény to Kisfái via Traktor.

Ping-conf Day 2

A második nap a Ping-conferencián hozta az első nap színvonalát. Yevgeniy Brikman izgalmas demóval egybekötött előadása mindenkit lázba hozott korán reggel. Prezentációját azzal kezdte, hogy elmesélte, hogy másfél éve használják a Play Frameworköt LinkedInnél több mint 60 projekthez nagy sikerrel. Megnevezett két problémás területet tapasztalataik alapján:

• Nehezen kezelhető komplexitás (szerintem ez nem feltétlen csak Playben jött volna elő az ő esetükben)
• Borzalmas teljesítmény

Az első ponttal kapcsolatban megmutatta, hogy a LinkedIn oldalán rengeteg apró kis tartalmi rész van, amik ráadásul eltérnek a különböző felhasználó típusonként (újonc, prémium, etc), ezért nem lehetséges egyetlen kontrollerbe tenni a logikát (hecc kedvéért kipróbálták, és a fordító elszállt a scala fájl méretére hivatkozva). A megoldás, amit használnak, hogy minden egyes kis doboz önállóan is él, tehát van saját kontrollere, nézete, meg minden ami kell. Nagy előnye a módszernek, hogy nagyon könnyen tesztelhető, továbbá nagyon könnyen skálázható vízszintesen, ugyanis az egyes modulok szétoszthatóak több szerver között. A publikus oldalak kontrollereiben pedig egyszerűen legenerálják az egyes részeket, és összefűzik a kiszolgálandó HTMLbe. A technika hátulütője, hogy a statikus erőforrásokat a kontrollerben kell kezelni (szerk,. megj. illetve nem lehet ezen erőforrások tömörítését, egybefűzését valamilyen pluginnel leegyszerűsíteni).

Ezt a fragmentációs technikát választva az alábbiak igényelnek külön odafigyelést:

• Cookiek esetén szükségük van egy metódusra, ami a darabkák headerjeiből összefűzte a Cookie bejegyzéseket
• Statikus erőforrások beillesztésére szintén a headerbe kell tenni  X.CSS ill. X-JS bejegyzéseket, amit aztán összefűznek, deduplikálják, és a HTML headbe tesznek (erről később, hogy miért pont oda)
• Hibakezeléshez a legfelső réteg kiolvassa a darabkák státsz-kódját

Miután végigmentünk a komplexitás témakörén, áttértünk a teljesítmény problémákra. Első lépésben átalakította az alkalmazást, hogy a hagyományos String alapú HTML generálás helyett streamelje a kimenetet. A Scala nyelvben vannak natív eszközök, mindössze egy custom render templatet kellett bekötni a Playbe (az alapértelmezett StringBuildert használ), és az Enumerator, Future párossal megoldotta, hogy a szerver az előállított HTMLt azonnal küldte is a kliensnek. Akit érdekel részletesebben a dolog itt tud utánaolvasni. Ez azért növeli a teljesítményt, mert a HTML headben elhelyezett erőforrások letöltésével nem kell a kliensnek megvárnia, amíg minden kis darabka összeáll. Következő trükk, amivel tovább fokozta a teljesítmény és a felhasználói élményt eléggé leleményes. A darabkák kimenetét a HTML bodyn kívülre, egy script type="text/html-stream" elembe tette, majd JavaScripttel a helyükre illesztette, amikor végzett velük a szerver. A módszer előnye, hogy nincs szükség darabonként egy AJAX kérésre, így tehermentesíthető a szerver, mégis a felhasználó azonnal kap választ a kérésére. Amire érdemes odafigyelni, hogy a hirtelen a helyükre kerülő elemek miatt ugrálhat az oldal, megzavarva ezzel az éppen kattintani vágyó felhasználót, valamint a headerek már a kérés elején kiküldésre kerülnek, ezért azokat a kérés további részében nem lehet változtatni. Nehezebb az oldalt tesztelni, és SEO problémák is felmerülhetnek. A demóalkalmazás egyébként elérhető githubon.

A következő prezentációt James Proper tartotta, aki szintén a teljesítmény-optimalizációt választotta témájául. Az első dolog, amit szép teljesítmény-tesztekkel bemutatott, hogy az aszinkron nem egyenlő a gyorsasággal, sőt! Egy egyszerű példán keresztül bebizonyította, hogy az aszinkron kérés mennyivel tud lassabb lenni a sok context switch miatt, amik elveszik a drága processzoridőt. Folytatásként több módszert is bemutatott, amivel növelhető a teljesítmény, és elkerülhető a rengeteg context switch.

• Az első javaslata az volt, hogy használjuk a Scala beépített ExecutionContextjét, ami a Java Fork/Join osztályaira épül. Bővebben
• Második lehetőségként azt ajánlotta, hogy ne használjuk az első pontban említett contextet, hanem váltsunk ImmediateExecutionContextre

Általános jó tanácsként említette, hogy sose használjunk blokkolt erőforrásokat ExecutionContextek használatakor, ha mégis szükségünk van rá, akkor inkább konfiguráljunk menetirányítókat típusonként (na jó ezt senki sem érteni :) "Configure dispatcher per type"). Mivel az Akka tud ilyet, és a Playben van beépített Akka, adja magát, hogy azt érdemes használni.

Kiemelte azt is, hogy nagyon nagyon nagyon nagy weboldalaknál lehet teljesítmény-növekedést elérni, ha a routingot több különálló fájlba tesszük, és egy custom router megírásával csak a megfelelő fájlt dolgozzuk fel, amikor a kérés beérkezik. A gyorsulás oka, hogy a  Play regexpeket illeszt az URI-re, és ha több száz illesztést kell csinálnia, mire a route fájl végén megtalálja a keresett bejegyzést, az felesleges processzor-terhelést jelent. A futtatott benchmarkoknál pár százalékos teljesítmény-növekedés volt mérhető.


A teljesítmény optimalizációs témaköröket elhagyva a következő előadást ismét  Julien Tournay és Pascal Voitot tartotta, és témája a Scalaz-Stream volt. A prezentáció lényege pár mondatban összegfoglalva annyi, hogy miként lehet streamek fogadására lecserélni a jelenleg is működő Iteratee megoldást Playben erre, az egyébként még fejlesztés alatt álló, megoldásra. A prezentációban egyetemi előadásokra emlékeztető részletességgel  mesélték el többek között, hogy miért is jó ez az újfajta megközelítés, hogy milyen problémákba ütközhetünk a HTTP, és a WebSocket eltérősége miatt, stb.


Ebédszünet előtt még Adam Evans és Asher Glynn tartott egy (számomra nem annyira érdekes) előadást arról, hogy milyen tapasztalataik voltak, amikor a BBC gyerekeknek szóló PHP-s megoldását lecserélték Play frameworkre. Architekturálisan volt egy PHP frontendjük, ami a BBC Java-s backendjéhez fordult adatokért. Ami miatt a Playre esett a választásuk:

• Mert Scala, ezt nem is fejtették ki bővebben
• Full stack web-framework
• Reaktív, könnyű benne non blocking szolgáltatásokat készíteni és hívni
• Typed safe template rendszer eléggé awesome
• Sok vállalat használja, és zizeg az egész

Miután kiválasztották a keretrendszert csináltak egy pilot projektet, ami nagyon jól sikerült, a PHP fejlesztőknek könnyű volt átszokni Scalára, és mindenki boldog.

A sponsor pitch után következő előadás igazán érdekes volt. Grant Klopper a The Guardian hírportál frontendjét kiszolgáló alkalmazásról beszélt, milyen problémákkal kell szembenézniük, milyen megoldásaik vannak, stb. Elmesélte, hogy naponta három Budapestnyi ember látogatja meg a weboldalukat, átlagosan mindegyik megnéz három oldalt. 900 kérést kell kiszolgálniuk minden másodpercben. A rendszer teljesen nyílt forrású, és elérhető a githubon, sőt nem csak elérhető, hanem konkrétan onnan buildelik a live rendszert. A teljes frontend letölthető egyetlen futtatható jar fájlként, és elindítható a java -jar frontend-artifact.jar paranccsal. A rengeteg kérés miatt mindent cachelnek, és a rendszer garantálja, hogy nem töltődik be sohasem ugyanaz a tartalom kétszer, a második kérés mindenféleképpen cache hit lesz. Beszélt továbbá még a deployment folyamatukról is, ami abból áll, hogy elindítanak három új verziót, regisztrálják őket a load balancerbe, majd lekapcsolják a régi hármat. Előadás után külön odamentem hozzá, és rákérdeztem miként kezelik azt a szituációt, amikor az összes node fut, de valami kódváltozás miatt a cachelt elem is megváltozik, és a régi verzióknak még a régit kell kiszolgálniuk, az újaknak pedig már az újakat. Dolgozom elosztott szinkron cachel, és ez nálunk bizony okoz problémát és fejfájást, éppen ezért mi kikapcsoljuk a cachet egészen addig, amíg minden nodeon az új verziójú szoftver nem fut. A válasz egész egyszerűen annyi volt, hogy sehogy. Azt mondta, hogy olyan rövid ideig fordulhat elő ez az eset, és a magas kérésszám miatt a statisztikában meg sem jeleni az a pár hibás kérés.

Utolsó előtti előadást Tobias Neef tartotta a kontrollerek absztraktálásának lehetőségeiről a Playben, hogy a lehető legjobban elkerülhessük a kódismétlést. Pár pontban összeszedve a lényeg:

• Készítsünk Actionönek az Actionjeinkhez
• Használjunk ActionBuildereket
• Compozáljuk actionjeinket (szép kifejezés)
• Használjuk a Filtereket
• Egyesítsük a hibakezelési és loggolási részeket

A konferencia Johan Andrén előadásával zárult, aki a különböző aszinkron lehetőségekről beszélt Scala és Java platformokon. Először a problémára világított rá Johan, miszerint az általunk írt kód a kiszolgálási idő nagyon kicsi részében van végrehajtási fázisban, és az idő nagy részét különböző erőforrásokra való várakozással tölti. Tegyük fel, hogy ha van 20 adatbázis, és 200 kiszolgáló szálunk, akkor a 200 + 1-edik adatbázist nem használó szál nem lesz kiszolgálva. A problémára háromféle megoldást mutatott be.

Az első, használjuk bátran a Future és Promise osztályokat (részletes információt itt találsz). Hátrányának azt nevezte meg, hogy amikor több rétegen keresztül dobáljuk a Future osztályokat, a rétegek között állandó jelleggel konvertálni kell azokat. Például egyik szerviz visszatér a userek listájával, de a hívónak JSON listát kell visszaadni, amit az őt hívónak HTML kóddá kell alakítani, stb. stb. Javaslata szerint az alábbi esetekben érdemes ezt a megoldás választanunk:

• Amikor más szolgáltatásokkal kommunikálunk
• Párhuzamos végrehajtásra, amikor a szálak teljesen elkülöníthetők egymástól
• Egyszerű háttérszolgáltatások implementálásakor

Utóbbi esetre nem tenném a nyakam, Tomcat esetén mivel a kiszolgáló szál referenciát őriz a Future objektumra szépen bevárja annak végeredményét, szóval én nem javaslom ezt a kombinációt.

Következő lehetőségként az Akka Actorsokat vetette fel, segítségével esemény vezérelt programozást tudunk megvalósítani. Az eljárás lényege, hogy van egy bemenő sor, amibe be tudják a kiszolgáló-szálak küldeni a kéréseket, és a rendszer egy szálon szépen sorban végrehajtja azokat. A technika nem sebességéről híres, blokkolja a többi futó szálat, de cserébe jól skálázható. Mikor használjuk:

• Amikor állapotokra van szükségünk
• Adat streamelés
• Esemény vezérelt programozásra
• Háttérfolyamatok végzésére

Utolsó lehetősségként a Iteratee-k használatát mutatta be. Sajnos ennek nincs Java-s megfelelője, a Chunks API tud valami hasonlót, de képességei messze elmaradnak a Scala natív megoldásától. A dolog lényege, hogy pici szeletekben adható át a feldolgozásnak az adat (Enumerators), és a rendszer lényegében reagál az adatra, és várja a következő darabkát.

• Enumerator[A] -> Iteratee[B, R]
• Enumerator[A] -> Enumeratee[A, B] -> Iteratee[C, R]

Lényeges külömbség a hagyományos feldolgozással szemben, hogy az első hibás darabka esetén már lehet kezelni a hibát, nem kell a teljes adatfolyamot megvárni.

Mikor használjuk:

• Streamel adatok esetén

A konferencia végére igencsak megfogyatkozott a létszám, láthatóan mindenki kellőképpen elfáradt. Meg kell hagyni, hogy elég tartalmasak voltak az előadások, szóval le a kalappal a szervezők előtt, nem bízták a véletlenre a mentális zombiságunkat. Gratula ezúton is.