Hadoop schaalt tot tientallen Petabytes verspreid over duizenden systemen

Internet-raamwerk voor de grote jongens

Het zijn gouden tijden voor techneuten. Iedereen met voldoende kennis en vaardigheden, genoeg ambitie en niet te veel scrupules kan een prachtige internet-onderneming van de grond trekken. Maar voor wie groot denkt, is een standaard server-park niet toereikend. Bedrijven als Google, Twitter en Facebook bewaren en bewerken enorme hoeveelheden gegevens. Zij gebruiken daarvoor meestal een parallel platform als Hadoop.

Net voor de jaarwisseling verscheen versie 1.0. Maar begin vorig jaar won het platform al de eerste prijs bij de MediaGuardian Innovation Awards. De jury benadrukte daarbij met name het belang voor innovatieve bedrijven als Twitter en Facebook.

Eén miljoen servers

Hadoop is het platform voor de internet-ondernemer die in het groot denkt. Als je werkt aan de volgende Google, Twitter of Facebook, dan heb je heel veel verwerkingskracht, geheugen en opslag nodig, veel meer dan een batterij grote servers je kan bieden. Gelukkig hebben de meeste internet-diensten genoeg aan horizontale schaalbaarheid. Dat wil zeggen dat de prestaties en de capaciteit min of meer lineair toenemen met het aantal servers dat je neerzet. Vaak zijn dit juist kleine, goedkope systemen (commodity hardware), die makkelijk en snel vervangen kunnen worden.

Zo gebruikt Google in zijn datacenters losse moederborden en harddisks, die in zelf ontworpen rekken worden geschoven. Als je kijkt naar de omvang van hun infrastructuur — naar verluidt hebben ze met ongeveer een miljoen systemen twee procent van alle servers ter wereld in bezit — dan betaalt zo'n eigen ontwikkeling zich dubbel en dwars terug.

Andere grootverbruikers zijn bedrijven als Akamai, Amazon, Intel, Microsoft en Rackspace, zij het dat hun infrastructuur toch een orde grootte kleiner is.

Google-technologie

Niet voor niets hebben de ontwikkelaars van Hadoop zich laten leiden door dezelfde technologieën die ook aan de basis van Google stonden: MapReduce en het Google File System. Die eerste is een methode om parallelle problemen over een (heterogeen) cluster te verdelen. Voor wie bekend is met algoritmen: MapReduce lijkt op een gedistribueerde, functionele vorm van backtracking en recursie.

GFS is het bijbehorende, gedistribueerde bestandssysteem. Daarbij worden files in stukken geknipt (FileSplits) en over verschillende systemen verdeeld. Door die stukken ook nog meerdere malen te copiëren, wordt niet alleen voor redundancy gezorgd, maar kunnen bevragingen op hetzelfde bestand worden geparallelliseerd (throughput computing).

Schaalbaarheid

De eerste code voor Hadoop is meer dan tien jaar geleden geschreven door Doug Cutting. Hij werkte destijds aan een open source search engine gebaseerd op Java. Nutch verzorgde het crawler-gedeelte, Lucene de indexering en search-afhandeling. Toen Google in 2004 op de Usenix conferentie een paper over hun MapReduce algoritme publiceerde, gebruikte Cutting dat om de schaalbaarheid van Lucene te verbeteren. Het resultaat was Hadoop, dat inmiddels, net als Lucene en Nutch, is ondergebracht bij de Apache Software Foundation (ASF), waar Cutting op dit moment weer voorzitter van is.

Location awareness

Hoewel Hadoop in eerste instantie is ontwikkeld voor de Nutch/Lucene search engine, is het een algemeen raamwerk voor het beheer van een netwerk van nodes (Common) en het uitvoeren van gedistribueerde opdrachten daarop (MapReduce). Om de infrastructuur zowel robuust als efficiënt te maken, moet de software weten welke nodes zich in welke rekken bevinden en welke nodes netwerk-technisch gesproken dicht bij elkaar liggen (location awareness). Zo kunnen de redundante stukken data over verschillende rekken worden verspreid, zodat gegevens wel beschikbaar blijven als de stroomvoorziening of een netwerk-switch uitvalt. Tegelijkertijd wil je de worker nodes die gegevens laten bewerken die zich al op hetzelfde systeem bevinden of anders op een data node die op dezelfde switch is aangesloten. De informatie over de onderlinge onafhankelijkheid van verschillende nodes is dus van belang voor het opslag-gedeelte, terwijl de nabijheid belangrijk is bij het uitvoeren van de opdrachten.

Hadoop gebruikt daarvoor verschillende soorten nodes. Er is minstens één master node (NameNode), die fungeert als ingang voor het file system, bestaande uit een heleboel DataNodes. De JobTracker is verantwoordelijk voor de distributie van opdrachten (MapReduce) over de compute nodes (TaskTrackers). Meestal zijn de data nodes met een compute node gecombineerd in één worker node (slave nodes). De nodes communiceren tenslotte met elkaar via RPC over beveiligde ssh-verbindingen.

HDFS

Met de software suite wordt het Hadoop Distributed File System (HDFS) meegeleverd, dat de bestanden in blokken van 64 Mbyte over de data nodes verdeelt. Snapshots en backups van het cluster kunnen worden gemaakt door de huidige index-tabellen van de master node naar een Checkpoint of Backup node te schrijven. Daarnaast kan met behulp van Federation een uitgevallen master worden opgevangen. Deze recente toevoeging aan HDFS creëert meerdere alternatieve ingangen (NameNodes) op het file system. Daarmee wordt een veelbesproken single-point-of-failure verholpen. Checkpoints doen hetzelfde voor de JobTracker, zodat niet al het werk verloren is als deze crasht.

Om gegevens voor bewerking te laden en de uitkomsten daarna weer terug te halen, vragen clients eerst bij de NameNode op de master waar bepaalde gegevens zich in het bestandssysteem bevinden. De informatie zelf wordt vervolgens direct met de DataNodes uitgewisseld.

Maar er is ook een ingang voor een virtueel bestandssysteem beschikbaar. Andere toegangsmogelijkheden zijn FTP en HTTP.

Andere bestandssystemen

Maar Hadoop kan ook met andere gedistribueerde bestandssystemen dan HDFS overweg. De scheduler is in principe onafhankelijk van het onderliggende file system. De software bevat een API waarop drivers voor alternatieve bestandssystemen ontwikkeld kunnen worden. Wel is het voor de prestaties en efficiëntie van belang dat het file system location awareness ondersteund.

Naast HDFS is bijvoorbeeld ook CloudStore beschikbaar. Deze software is geschreven in C++ en biedt vergelijkbare functionaliteit als HDFS. Daarnaast kan Hadoop worden gekoppeld met de S3 cloud storage (Simple Storage Service) van Amazon. Andere ondersteunde bestandssystemen zijn IBM GPFS (General Parallel File System), IBRIX Fusion van HP, en MapR.

Big Data

Hadoop schaalt op tot duizenden nodes en tientallen Petabytes aan data. Voor dergelijke omgevingen, waarin de hoeveelheid informatie zo groot wordt dat deze niet meer op de traditionele manier te behappen is, wordt de term Big Data gebruikt. Al die gegevens kunnen niet langer in een (relationele) database worden opgeslagen, maar vereisen systemen en toolsets die massief parallel en horizontaal schaalbaar zijn. Toepassingen liggen in de wetenschap en in de zakelijke en financiële wereld, maar vooral aan de achterkant van die grote, wereldwijde internet-portals.

Inmiddels is er dan ook een hele zwik uitbreidingen gebouwd op Hadoop/HDFS beschikbaar:

• Cassandra: gedistribueerde database met replicatie
• Chukwa: log-analyse
• Dumbo: een Python API voor Hadoop
• Hama: BSP (Bulk Synchronous Parallel) library voor massief parallelle, wetenschappelijke berekeningen
• HBase: gedistribueerde database, vergelijkbaar met Google's Bigtable; ook geschikt voor het schrijven van records, in tegenstelling tot HDFS dat is geoptimaliseerd voor lezen
• Hive: data warehouding, analyses en queries met behulp van een eigen SQL-taal
• Hypertable: een gedistribueerde database geschreven in C++
• Mahout: aanbevelingen, groeperen, classificeren en datamining
• Pig: data-analyse met behulp van een eigen specificatie-taal
• Zookeeper: configuratie-management voor clusters

Installatie

De code voor Hadoop wordt gepubliceerd onder de Apache licentie. Voor de installatie zijn Java versie 1.6 en ssh(d) nodig. Hoewel je meestal Linux zult gebruiken, kan Hadoop ook op Windows voorzien van de Cygwin tools worden gedraaid. Voor een productiesysteem wordt dat echter afgeraden. Andere platforms waarop men Hadoop werkend heeft gekregen zijn BSD, OpenSolaris en OS/X.

Hadoop kan in drie verschillende modes draaien: Standalone, Pseudo-Distributed voor het draaien van een compleet netwerk op een enkel systeem, en Fully-Distributed voor een volledig gedistribueerd cluster in productie. De installatie en configuratie voor die eerste twee zijn eenvoudig: een kwestie van de software neerzetten en wat XML-bestanden aanmaken.

De cluster setup is ingewikkelder. Daarvoor moet om te beginnen de Hadoop stack op elke node worden geïnstalleerd. Daarnaast zijn er flink wat configuratie-opties. Tenslotte moet natuurlijk een lijst met adressen en rack-nummers van de slave nodes worden aangemaakt. Een dergelijke installatie heeft vanzelfsprekend meer voeten in de aarde dan een test-opstelling maar is toch niet overdreven ingewikkeld. Ook voor een goede hobbyist bekend met Linux en de hier uitgelegde concepten is het prima te doen.

MapReduce

De echte waarde van het cluster zit 'm natuurlijk in de bewerkingen op de gegevens die zich op het gedistribueerde bestandssysteem bevinden. Zo'n klus leg je vast in twee Java-functies: Map en Reduce. Die eerste voert een opdracht uit op alle data-blokken op een systeem, en dat tegelijkertijd op alle worker nodes. De JobTracker op de master node verzamelt vervolgens de deelresultaten van de TaskTrackers op de afzonderlijke nodes, terwijl de Reduce-functie die verwerkt tot een eindresultaat.

Hoewel het Hadoop framework in Java is geschreven, is dat niet noodzakelijk het geval voor de Map- en Reduce-functies. Je kunt de slave nodes ook C++-functies (via Hadoop Pipes) of shell commando's laten uitvoeren (via Hadoop Streaming).

Voor Hadoop-programmeurs is een plugin voor Eclipse beschikbaar, te gebruiken voor zowel Linux- als Windows-omgevingen.

Van start

Om snel van start te gaan, kun je de Hadoop Demo VM van Cloudera proberen, een VMware image met daarop CentOS, Hadoop, Hive en Pig. Een andere mogelijkheid is de OpenSolaris Hadoop LiveCD van Sun (tegenwoordig Oracle). Het image is nog steeds te downloaden, maar Oracle onderhoudt OpenSolaris al tijden niet meer. Voorheen draaide Hadoop ook op de Sun Cloud, maar ook die is niet meer beschikbaar.

Voor het eerste productie-systeem lijkt de Hadoop-dienst van Amazon een goede. Amazon Elastic MapReduce is gebaseerd op EC2 en S3. Daarmee heb je de volledige flexibiliteit en schaalbaarheid van een cloud-oplossing. Bovendien voorkom je hiermee de afstand tussen de compute nodes en de data nodes die je creëert als je je eigen Hadoop-systemen zou combineren met alleen S3-opslag. In dit geval kun je dus beter de complete dienst afnemen.

Hadoop-leveranciers

Ga je echt groeien, dan is een eigen infrastructuur natuurlijk goedkoper. Cloudera's Distribution Including Apache Hadoop (CDN) biedt kant-en-klare installatie-pakketten voor Debian/Ubuntu, RHEL/CentOS/Fedora en SuSE. Ook MapR levert een distributie voor RHEL/CentOS en Ubuntu. Na registratie op hun site kun je een demo in de vorm van een VMware image downloaden. MapR werkt hierbij samen met EMC, dat hun software stack weer onder de naam Greenplum HD Enterprise Edition verkoopt.

Voor professionele ondersteuning (anders dan vanuit de community) kun je bij de zojuist genoemde bedrijven terecht. Maar ook bij Hortonworks of traditionele leveranciers als Google (Google App Engine), HP, IBM (BigInsights), Oracle (Big Data Appliance, in samenwerking met Cloudera) en SGI (ook in samenwerking met Cloudera).

Daarnaast zijn er diverse leveranciers die zich hebben gespecialiseerd in specifieke toepassingen of specifieke markten.

Inspiratie

De twee grootste gebruikers van Hadoop zijn Facebook en Yahoo!. Die eerste heeft een groot cluster met tientallen Petabytes aan data staan en gebruikt dat onder andere voor het draaien van analyses. Yahoo! gebruikt Hadoop voor de indexering van zijn search engine. Daarnaast is het de grootste contributor voor de code en ook de toenmalige werkgever van Cutting. Andere bekende namen op de lijst van Hadoop-gebruikers zijn Adobe, Alibaba (analyses), Amazon/A9 (doorzoeken producten), AOL (analyses), eBay (zoek-optimalisatie en analyses), IBM (Blue Cloud dienst), foursquare, Last.fm (analyses), LinkedIn (People You May Know), Netflix, de New York Times (voor beeldbewerking), Quantcast, Rackspace (log processing), SARA, StumbleUpon (aanbevelingen), Twitter (grootverbruiker, voor opslag en verwerking van tweets) en Wikipedia.

Tenslotte nog een laatste tip voor wie door dit alles zo geïnspireerd is geraakt dat hij zelf met Hadoop aan de slag wil. Dan is Sector/Sphere een alternatief dat je zeker ook even moet bekijken. Waar HDFS de bestanden ophakt in blokken, werkt dit platform met de originele files. Bovendien is Sector gemaakt om over WAN-verbindingen te lopen. Daarvoor is een eigen netwerk-protocol op UDP ontwikkeld: UDT (UDP-based Data Transfer). Sector/Sphere is geschreven in C++ en beschikbaar voor Linux en Windows. Behalve dat de parallelle engine Sphere flexibeler lijkt dan MapReduce, is deze volgens de ontwikkelaars twee tot vier keer sneller dan Hadoop.