Hadoop

Els que estigueu posats en el món de l'informàtica distribuïda o bé d'alt rendiment o bé, simplement que esteu una mica a l'aguait de les novetats informàtiques potser us sonarà Hadoop.

Hadoop és una plataforma de software que bàsicament permet escriure programes que processin unes grans quantitats de dades. Grans grans. El projecte ja té diversos anys i alguns dels seus components s'han separat en subprojectes:

• Hadoop Core: Plataforma de Map Reduce + Sistema de fitxers distribuit.
• HBase: Una base de dades distribuïda i escalable.
• Pig: Nivell d'alt llenguatge per aplicacions que utilitzen Hadoop-core.
• ZooKeeper: Sistema de coordinació per Hadoop.
• Hive: Una espècie de SQL fer queries a les dades emmagatzemades en Hadoop (moltes dades, moltes moltes)

És un projecte Open Source utilitzat en producció (molt important) en diverses empreses, tal com podeu veure aquí. A més, té una comunitat d'usuaris / desenvolupadors molt activa tal com podreu veure si doneu un tomb pel seu tracker

En aquest post parlaré exclusivament del projecte principal, Hadoop Core.

Les característiques que ells destaquen en la seva pàgina són les següents: 

• Escalabilitat: Pot manejar grans quantitats de dades (petabytes) i té escalabilitat horitzontal, és a dir, si necessites més potència o més espai, simplement afegeixes màquines.
• Econòmic: És Open Source (llicència Apache) i treballa amb qualsevol tipus de màquines, no és necessari que siguin servidors d'alta gamma. A més, és una plataforma de software que suposa que els errors de hardware són usuals i per tant, sap reaccionar. També cal destacar que està preparat per còrrer al Amazon EC2
• Segur: Sempre es mantenen diverses còpies de les dades repartides en diverses màquines, així ens estalviem desgràcies quan una màquina decideix morir sense avisar. 

Hadoop és capaç de processar grans quantitats de dades perquè utilitza Map Reduce. Map Reduce és un model de programació pensat, o més bé, readaptat, per procés de moltes dades. Amb això voldria aclarar que de fet, només serveix per això. Sembla ser que hi ha certa moda en utilitzar Map Reduce en alguna de les seves implementacions per moltes coses. Si no es té una entrada de com a mínim gigues de dades probablement no sigui el que es necessita.

Bé, els causants de tota aquesta febre MR (Map Reduce) van ser Google quan van publicar un article on explicaven què era, com es programava i com ho utilitzaven en producció a diari en els seus clústers. A partir d'aquí van començar a aparèixer implementacions d'aquest model de programació en diversos llenguatges, una de les més importants, si no és la que més és sens dubte Hadoop. 

En resum, MR i per tant Hadoop treballen de la següent manera (molt per sobre):

Tenim les dades al sistema de fitxers distribuit que comparteixen totes les màquines que conformen el nostre clúster. Tenim un programa que implementa MR mitjançant Hadoop, això bàsicament vol dir que hem programat un programa semblant a aquesta, el qual té una classe Mapper, amb un mètode map i una classe Reducer amb un mètode reduce. 

Hadoop parteix el clúster en mappers i reducers, una màquina pot ser mapper i reducer a l'hora, això es pot definir en la configuració del sistema. Les màquines que fan de mappers agafen les dades del sistema de fitxers distribuit i li apliquen la funció map. Així de bon principi ja tenim tot de màquines que estan llegint les dades d'entrada de forma paral·lela. 

MR té uns requeriments per les dades. Bàsicament han d'esser de la forma. Clau -> Valor, on la clau i el valor poden ser qualsevol cosa. Per tant, un cas típic seria un log d'apache:

127.0.0.1 - - [31/Jul/2008:00:37:04 +0100] "GET /~marc/images/imatge1.jpg HTTP/1.1" 200 16624
127.0.0.2 - - [31/Jul/2008:00:37:05 +0100] "GET /~marc/images/imatge2.jpg HTTP/1.1" 200 16624
127.0.0.3 - - [31/Jul/2008:00:37:06 +0100] "GET /~marc/images/imatge3.jpg HTTP/1.1" 200 16624
127.0.0.1 - - [31/Jul/2008:00:37:04 +0100] "GET /~marc/images/imatge3.jpg HTTP/1.1" 200 16624

Per exemple en aquest programa ens interessa saber quines IPs han agafat quines imatges. Per tant podriem dir-li al mapper que la clau de sortida sigui la imatge, i que el valor de la clau sigui la IP: 

Clau   Valor
imatge1.jpg  127.0.0.1 
imatge2.jpg 127.0.0.2 
imatge3.jpg 127.0.0.3 
imatge3.jpg 127.0.0.1

Ok, per cada Clau/Valor que el mapa troba l'envia a un reducer. És important notar que no associa res de res, en aquest cas enviarà una clau/valor per cada linia de fitxer. Aquesta és una part molt interessant ja que el mètode map simplement invoca una funció amb la clau i el valor com a paràmetres i se n'oblida. Aquests valors són emmagatzemats "màgicament" al sistema de fitxers distribuit de forma totalment transparent fins que tots els mappers han acabat. Això vol dir que els reducers no comencen a processar les dades fins que els mappers estiguin (en veritat si que comencen en certa manera, però no és important ara). Hadoop té un particionador de manera que, atenció que això és important, cada reducer (cada instància d'un reducer) rep una i només una clau amb una llista dels valors: Això és una mica espinós per entendre, exemple:

Clau 1 / Valor A
 Clau 2 / Valor B
 Clau 3 / Valor A
 Clau 2 / Valor A
 Clau 3 / Valor C

I el reducer rebrà:

Reducer1 : Clau 1 amb una llista que contindrà: Valor A
Reducer2 : Clau 2 amb una llista que contindrà: Valor B, Valor A
Reducer3 : Clau 3 amb una llista que contindrà: Valor A, Valor C 

Per tant hi haurà tants reducers com claus tingui la sortida el mapper. En l'exemple anterior es tradueix a: 

imatge1.jpg 127.0.0.1
imatge2.jpg 127.0.0.2
imatge3.jpg 127.0.0.3, 127.0.0.1

Llavors lògicament el reducer pot fer el que vulgui amb aquestes dades. L'exemple aquest és extremadament senzill, però pensem que un map i un reduce complex poden fer les mil i una bestieses i si estem parlant d'un fitxer de log complert d'apache d'alguna pàgina important pot ser molt i molt gros.

Un cop el reduce està també escriu els resultats en forma clau valor. Aquests resultats són guardats al sistema de fitxers distribuit. 

Podeu trobar molta més informació sobre Map/Reduce + Hadoop al tutorial de hadoop. 

Bé, en pròxims articles entraré més en detalls sobre Hadoop i probablement publicaré snippets de codi, amb problemes que em vaig trobant al dia a dia i alguns números.

CouchDB i Key-Value Stores.

Avui he estat a una presentació de Jan Lehnard sobre CouchDB. Podeu trobar les slides aquí.

CouchDB és una key-value store, i aquest és justament el primer post sobre aquest tema. En vindran molts més ja que és un tema molt extens i molt interessant, si més no per mi. Bàsicament podriem dir que una key-value store és una base de dades molt senzilla distribuïda per la xarxa, també hi ha qui ho enten com una taula de hash gegantina distribuïda en diversos ordinadors. I molt més pensada per a l'eficiència dels gets/sets que a per la complexitat que poden tenir altres bases de dades com Postgres o Mysql.

Una de les coses que m'ha fet pensar més de la xerrada és la quantitat de key-value stores que hi ha actualment. La majoria són bastant noves i tot i que totes serveixen pel mateix, emmagatzemar i obtenir certs valors, els seus dissenys són totalment diferents i fan que cadascuna tingui la seva sortida de mercat.

Aquí podeu trobar un post molt complert i interessant sobre key-value que va fer l'RJ (CTO de last.fm)

Unes pinzellades per a qui li faci mandra llegir-se tot l'article o li faci cosa perquè està en anglés:

Les principals diferències de disseny de les key-value stores que convé mirar abans d'adoptar-ne una, o bé que s'han de tenir en compte són:

• Tolerància a fallades: Les key-value stores normalment estan en màquines dedicades, i més d'una màquina, per tant diem que tenim 5 màquines servint l'informació. Què passa si una de les màquines s'espatlla de cop?. Depenent de la tolerància a fallades de la key-value store pot ser que no passi res, o que tinguem un problema gros. Entre els tipus de tolerància a fallades que són més normals destacaria:
  
  
  
  • Replicació: Tant senzill com tenir més d'una còpia de cada valor que es serveix. Si cau una màquina no passa res perquè hi haurà alguna altre màquina que també té aquest valor. Lògicament gastem més espai del necessari, però és el preu a pagar.
  
  
  • Particionat: No és una tècnica en si sola, sino que és una millora del mètode anterior, si els valors són grans (fitxers, mp3, imatges) podem partir el fitxer en diversos trossos i repartir-los (amb redundància o replicació) per diverses màquines.
    
  
  
• Persistència: Normalment aquests mecanismes tenen totes les dades a memòria RAM. Amb això ens estalviem tot el temps d'I/O a disc. Lògicament fa que quedi poca RAM per altres coses, però és el preu de la velocitat. Ara bé, de nou, què passa si hem d'apagar la màquina?. N'hi ha que guarden els valors en una base de dades relacional, ja sigui Mysql, Berkeley, etc. Així, al engegar agafen tota la informació de la base de dades i la posen a memòria. Treballen en memòria i finalment, quan s'apaga la màquina es guarda de nou a la base de dades. N'hi ha d'altres que ho serialitzen utilitzant llibreries natives dels llenguatges de programació en que estan implementades i fins i tot n'hi ha que _no_ guarden res. Quan s'apaguen es perd tot!, normalment són utlitzades com a caches.


• Accés: Molt bé, tenim molta informació, però com hi accedim?. N'hi ha que porten les seves pròpies llibreries en diversos llenguatges, n'hi ha que utilitzen HTTP (sempre REST, mai SOAP) i n'hi ha d'altres que ofereixen interficies de thrift o protocol buffers (hi haurà tot un post sobre aquests dos, no patiu, són mecanismes de serialització / RPC, el primer fet a facebook i el segon fet a google).


• Eficiència: Lògicament, s'ha de parlar d'eficiència. Tant en temps d'accés a les dades (quan tardes en fer un GET, quan tardes en fer un PUT?) com en l'eficiència d'ús de recursos que utilitzen les màquines per servir l'informació.
  

En pròxims posts entraré una mica més en detall en implementacions concretes i en les key-values stores que considero més interessants.

Bittorrent

Començo el nou blog amb un post sobre protocols, avui Bittorrent!

Suposo que tothom, més o menys coneix el Bittorrent, aquest "programa per descarregar pel·lícules". Doncs bé, bittorrent és el nom del protocol, també de l'empresa que el va fer i d'un programa que utilitza aquest protocol. Sí, realment poca imaginació.

Bàsicament és un protocol per distribuir fitxers, fa temps s'usa de forma popular, però últimament ha tingut bastant resó mediàtic pel judici que ha tingut lloc a Suècia contra la pàgina d'emmagatzemament de .torrents The Pirate Bay. Al final sembla ser que poca cosa han pogut fer perquè tal pàgina tant sols guardava els fitxers .torrent i només actuava de tracker, i això no és delicte oi ?, ara probablement algú es preguntarà què és un .torrent? què és un tracker ? ara anem a això!

Protocol

El gran avantate que té el bittorrent sobre l'HTTP clàssic és que quan hi ha diverses baixades del mateix fitxer de forma concurrent (diverses persones volen baixar-se el mateix fitxer alhora), aquests usuaris s'envien trossos del fitxer entre ells. Per tant, com més persones hi hagi baixant un fitxer millor.

Com funciona.

El funcionament de cara l'usuari és ben senzill.

Un usuari es baixa un fitxer amb extensió .torrent, és un fitxer petit, menys de 60ks. Tot seguit s'obre amb el client de bittorrent i el fitxer es comença a descarregar. Pim pam.

En resum i explicat d'una forma planera el que passa és el següent:

Quan l'usuari obre el fitxer .torrent amb el seu client, aquest contacta amb el tracker (la direcció del tracker està dins el fitxer .torrent) que és un programa instal·lat en un servidor d'internet, i s'identifica com un usuari que es vol baixar aquell fitxer en concret. Tot seguit el tracker contesta amb les IP's dels diferents peers (altres usuaris) que s'estan baixant el fitxer, i que per tant, en ténen trossos. Seguidament el client contacta amb aquesta llista d'usuaris i els hi comença a demanar trossos, més endavant aquest client també compartirà els trossos que ja té.

De tant en tant contactaran amb el tracker de nou per refrescar la llista d'usuaris connectats que comparteixen aquest fitxer.

Fitxer Torrent

El fitxer .torrent és un fitxer binari. Bàsicament és un diccionari amb els següents camps i valors:

• announce: La URL del tracker.
• info: És un altre diccionari que conté:
  
• name: Nom suggerit per guardar el fitxer o el directori.
  
• pieces length: Mida dels trossos en que es parteix el fitxer (o fitxers) en bytes. El protocol parteix el fitxer en trossos del mateix tamany, amb la única possible excepció de l'últim tros. Els trossos normalment són de 256Ks tot i que es veu que en versions del protocol anteriors els trossos eren de 1Mb.
  
• pieces: És un string, la mida del qual és múltiple de 20. Cada 20 caràcters representa el hashing SHA1 del tros que representa, per tant hi haurà tants subtrossos de 20 com trossos tingui el fitxer, i a cada posició 20*i , sient 0 <= i < #trossos hi ha el hashing per a aquell tros en concret.
• length/files: O bé hi ha length o bé hi ha files, però estrictament un (i només un) dels dos. En cas de que sigui length, significa que el que s'està baixant és un sol fitxer i en length és el tamany total d'aquest en bytes. En cas de que sigui files representa un conjunt de fitxers dins d'una estructura de directori.
  
• path: Una llista d'strings que corresponen a noms de subdirectoris, en cas de que existeixin.
  

Tracker

El tracker és el software que comunica els usuaris amb altres usuaris, hi ha diverses implementacions, una de les més famoses és opentracker, que és la que utilitzen entre d'altres The Pirate Bay i la Wikipedia. A destacar la seva llicència, beerware.

Quan un client fa un GET (demana informació) a un tracker, la comanda HTTP té els següents valors:

• info_hash: És un string de 20 bytes amb el hash SHA1 de l'apartat info del fitxer de metainfomració (aka .torrent)
  
• peer_id: Id de l'usuari que es vol baixar el fitxer, aquest id és totalment aleatori i és generat pel client cada vegada que comença a baixar-se un fitxer de nou.
  
• ip: (Paràmetre opcional) IP del client.
  
• port: És el port al qual escolta el client, normalment és el 6881, en cas de que estigui ocupat és el 6882 i anar pujant fins al 6889.
  
• uploaded: Els bytes que s'han pujat d'aquest fitxer. Això són els bytes que aquest client en particular ha enviat a altres peers.
  
• downloaded: Bytes que s'han baixat fins aquest moment.
  
• left: Número de bytes que falten per completar la transferència del fitxer. Cal notar que no té perquè ser el mateix que el resultat de la mida del fitxer - downloaded, ja que pot ser que hi hagi hagut una corrupció de dades o problemes de transferència, i que per tant algun tros de fitxer s'hagi de tornar a baixar.
  
• event: És l'estat del client, pot ser started, quan s'està començant a baixar des del començament, completed un cop té tot el fitxer o stopped, un cop es para de baixar.

La resposta del tracker al client també són diccionaris en binari. Pot ser que el tracker contesti amb una entrada de failure, amb un contingut que és un string llegible amb el missatge d'error. En cas de que tot hagi anat bé el diccionari conté dues claus, la primera interval, que són el número de segons en que el client hauria de tornar a fer una petició d'informació al tracker. La següent entrada serà peers, que conté una llista de id's, ips i ports dels peers que s'estan baixant el fitxer. En cas de que hi hagi algun aconteixement especial un client pot fer una requesta d'informació al tracker encara que no hagi passat el temps indicat per aquest.

Comunicació entre peers

La comunicació entre peers és simètrica. Els missatges que s'envien d'un peer a l'altre són del mateix format i les dades (del fitxer que s'està compartint) poden anar també en els dos sentits.

Els peers es refereixen als diferents trossos del fitxer compartit amb indexs, que són exactament els mateixos indexs que hi ha al fitxer de metainformació. Un cop un client s'ha baixat un tros del fitxer i n'ha comprovat l'integritat amb el mapa SHA1 ho comunica a la resta de peers.

Les connexions entre peer i peer contenen dos bits d'estat. El primer, choked (tapat), si o no, i el segon, interessat, si o no. El bit the choked indica que el client està tapat, això vol dir que no enviarà més informació fins que sigui "destapat".

Una transferència de dades entre dos peers té lloc quan un dels dos està interessat i l'altre no està tapat. En cas de que un client no estigui interessat en res que tingui un altre client (sempre que aquest últim estigui destapat) aquest ho ha de fer constar amb el bit de interessat a fals.

El protocol recomana que quan s'està enviant dades, els clients haurien de tenir encuades a memòria diverses peticions de trossos per tenir una bona rendiment TCP, per altra banda quan s'està enviant dades a un altre peer també haurien de tenir els trossos a enviar a memòria, i en cas de que s'anul·li per qualsevol motiu la transferència, probablement el client quedi tapat. aquests trossos es poden borrar fàcilment. Això com ja he dit són recomanacions del protocol, caldria mirar exactament com s'implementa en cada client.

Per tant, quan dos clients s'envien missatges tenim bàsicament que comencen amb un handshake seguit per un fluxe d'informació de mida prefixada.

El handshake comença amb el número 19 en decimal seguit de l'string 'BitTorrent protocol'. A partir d'aquí, tots els números seran enters codficats en 4 bytes en big-endian. Seguidament venen els 20 bytes de hash SHA1 de la taula info_value del fitxer de metainformació, en cas de que els dos clients no donguin exactament el mateix hash, es talla la comunicació entre ells. En cas de que tot vagi bé s'envien 20 bytes més amb l'identificació del clients, en cas de que aquesta informació (l'id del client amb qui s'està comunicant) no aparegui a la informació sobre clients que havia donat el tracker, es talla la comunicació.

Tot seguit tenim el que és la comunicació en si entre peers, els missatges amb mida de cos cero són pings per fer saber i mantenir l'estat dels peers i s'envien cada dos minuts. En cas de que el cos del missatge tingui més d'un byte llavors cal mirar quin tipus de missatge es tracte:

• 0 - Choke
• 1 - Unchoke
• 2 - Interested
• 3 - Not Interested
• 4 - Have: És un número que indica que s'ha acabat de transferir i comprovat la integritat d'un nou tros.
  
• 5 - Bitfield: És un mapa de bits on cadascun indica quins trossos de fitxer s'ha enviat.
• 6 - request: Bàsicament és una petició perquè li enviin un tros. De paràmetre té un index.
  
• 7 - piece: Les dades en si del fitxer.
  
• 8 - cancel: Demana per cancel·lar una transmissió.
  

A l'hora de baixar els trossos normalment els clients trien els trossos de forma aleatori, així s'aconsegueix que estranyament es donguin situacions d'inanició que són les quals en que hi ha alguns trossos del fitxer que només estiguin en un o poc clients, cosa que faria que en el cas que aquests paressin de pujar fitxers ningú podria tenir el fitxer complert. Això forma part del protocol oficial de Bittorrent, tot i que algunes implementacions intenten millorar aquest aspecte fent que els trossos menys continguts en global entre els clients siguin els que tenen més prioritat per ser enviats.

i bàsicament... és això :)

Jerga

Vinga va, una mica de paraulotes relacionades amb el tema:

• p2p: Peer 2 Peer, d'un a l'altre, d'un punt a l'altre, d'una persona a una altre, bla bla bla
  
• peer: Un punt (ordinador, màquina, persona...) que participa en una comunicació P2P
  
• seed: En el món bittorrent seeder és una persona que permet que d'altres puguin obtenir el fitxer un cop ell/a ja el té. Per tant, que no borra el fitxer una vegada l'ha baixat, sinó que el continua compartint.
  
• leecher: justament al revés que el seeder, una persona que no comparteix el fitxer, tant sols el baixa.
  

Links

• Especificació del protocol. Bastant ben explicat i bastant planera.
• Wikipedia Hi ha un gràfic molt interessant. Pago una cervesa a qui l'entengui!
  

Primer post i presentació.

Bé, començo el nou blog sobre informàtica distribuïda.

Com suposo ja sabreu n'hi ha molts de blogs d'aquesta temàtica per la xarxa. El que passa és que la majoria són en anglés.

Per tant m'he decidit en obrir-ne un en català.

Aquí aniré traduint els diversos articles i posts d'altres blogs que em semblin interessants i aniré fent-ne de propis, lògicament.

Bàsicament parlaré de protocols de comunicació, mecanismes de RPC, de serialització, de diverses eines i dissenys per arquitectures i sistemes distribuïts.

apa doncs.