Hadoop streaming avec PHP et extraction HIVE – part 2

Dans cette suite de billets consacrés à HADOOP, HIVE et Map Reduce, nous allons présenter «Hadoop streaming» et voir comment il est simple d’utiliser les fonctions natives de Map Reduce du framework «Hadoop» avec PHP.

En reprenant le schéma de la première partie consacré à l’introduction d’«hadoop» et à son installation, nous nous attarderons sur les services de «Logs», «App Engine», «HIVE» et «HDFS» de l’architecture. Pour avoir une idée de comment s’articule l’ensemble de ces services jusqu’à la visualisation des métriques marketing, je vous propose le plan suivant :

• collecte et centralisation des logs,
• manipulation des logs et utilisation de Map Reduce,
• introduction à HIVE,
• extraction de données.

Savoir écouter son marché pour collecter de l’information stratégique

Lorsque le business modèle d’une entreprise s’articule sur le web et pour le web, elle est capable de mettre en place une interaction «OneToOne» (Peppers & Rogers) avec l’ensemble de son marché !. En contre partie, elle doit être capable de traiter rapidement l’ensemble des signaux sociaux économiques au risque de voir disparaitre le ciment de son activité.

En fonction du secteur d’activité de l’entreprise et de sa réactivité face au marché, la volumétrie de l’entrepôt de données sera plus ou moins importante et qualifiée. Ainsi, l’infrastructure, définie dans la première partie, faisant référence à une entreprise de vente de nourriture pour animaux sur internet, aurait pu être celle d’une entreprise dans le secteur du jeu social sur Facebook.

Ces entreprises sont confrontées à une concurrence de plus en plus soutenue (nouveaux entrants ; services concurrents toujours plus novateurs ; prix) et doivent piloter leur activité stratégique de manière très fine. C’est pourquoi, la qualité de l’information captée devient un enjeu stratégique. L’entreprise doit non seulement être capable d’identifier l’information nécessaire au pilotage de son activité, et être en mesure de la traiter pour pouvoir agir sur ses forces productives. Elle doit être capable d’adapter son modèle économique le plus rapidement possible afin d’optimiser son rendement économique ou atteindre l’équilibre financier (cas des startup’s).

L’intéraction des utilisateurs (clients et non clients) sur le site web, la relation client avec le SAV, les évènements sociaux économiques extérieurs (crise économique, guerre, lobbies etc …), sont des sources de données que va pouvoir utiliser l’entreprise. Face à cette problématique de Big Data, en réussissant à analyser ces données, l’entreprise va pouvoir agir sur plusieurs leviers afin d’améliorer sa compétitivité et sa rentabilité économique :

• modifier l’«IHM» de son site et améliorer son attractivité,
• ajouter de nouvelles fonctionnalités attractive à ses jeux en ligne sur facebook,
• afficher de la publicité mieux ciblée pour améliorer ses revenus publicitaires,
• augmenter la fréquence des mises à jour de son logiciel SaaS afin d’éviter un turn over,
• programmer des «events» et développer un écosystème,
• etc…

Ces actions de modifications ou d’actions vont à leur tour influencer le comportement des utilisateurs. Ces changements comportementaux pourront être à leur tour analysés. L’entreprise rentre dans un marketing dynamique, capable de comprendre l’écosystème dans lequel elle évolue et d’impacter ce dernier à son profit. Le schéma ci-dessous résume ce cycle.

La difficulté n’est pas forcément de capter la donnée mais de choisir celle qui sera la plus pertinente :

• quelle donnée prendre ?,
• tout ?,
• un mixe ?,
• comment dois je la traiter?
• ….

Autant de questions que doit se poser l’entreprise au risque d’avoir une base de connaissance non utilisable. Dans la plupart des cas, on collecte les logs en provenance des serveurs web. Ces derniers sont le résultat d’une intéraction entre l’utilisateur et le système. Dans notre exemple d’infrastructure, nous avons choisi de centraliser les logs pour des raisons de commodité de traitement.

Traitement de log en PHP et utilisation de Map Reduce

Map Reduce est un principe de traitement qui distribue équitablement de gros traitement sur plusieurs plus petits, tous identiques (MAP). Les résultats des MAP sont regroupés dans un ultime traitement (REDUCE). Au sein du langage PHP, nous avons à notre disposition tout l’outillage pour réaliser des fonctions de MAP et de REDUCE. Par contre, nous n’avons pas de mécanisme pour homogénéiser la distribution de ces traitements. Certaines bases de données, comme «MongoDB», embarque ce principe de traitement. Il existe depuis peut un pont technique entre MongoDB et Hadoop. Le schéma, ci-dessous, représente les deux méthodes de traitement (Traditionnelle & Map Reduce).

Le framework Hadoop va nous fournir le socle technique pour lancer des jobs de Map et de Reduce. Il nous reste plus qu’à développer les scripts PHP et le «bootstrap» pour exécuter le workflow.

Constitution du fichier de log

Pour les besoins de notre exemple, vous pouvez télécharger le fichier de log ou le constituer à partir de vos propres logs. En fonction de votre problématique, vous pouvez envisager plusieurs traitements Map et Reduce pour obtenir la structure souhaitée de vos données. Il est important d’avoir à la fin du traitement, des données structurées et de même dimension. Cela est primordial pour pouvoir brancher HIVE sur notre entrepôt de donnée.

Je vous propose aussi de dupliquer le fichier de log afin de montrer comment se comporte HADOOP lorsqu’il rencontre plusieurs fichiers au sein d’une source. Cela pourra vous permettre de vous adapter en fonction de vos besoins.

#Duplication du fichier de log : on suppose que vous êtes à la racine de votre répertoire démo avec l’ensemble des scripts (php & sh).
mkdir log
for i in $( seq 1 15) ; do cp log_hadoop.log log/log${i}.log; done;
#un ls du répertoire log, vous listes les 15 fichiers de log

Script Shell

Le script Shell sera notre intermédiaire pour lancer le job de Map Reduce. Il servira aussi à préparer le HDFS pour la réception des données à traiter et le stockage du résultat. Nous utiliserons la librairie «Haddop Streaming» (hadoop-streaming-1.0.0.jar dans notre cas). Depuis peut, une mise à jour d’Hadoop est disponible (Hadoop-1.0.1). Le nom du «streamer» change avec le numéro de la version. Il aura pour rôle de :

• rediriger les flux issus de la lecture des fichiers de log sur l’entrée standard des «Mapper»,
• récupérer le flux de la sortie standard du «Mapper» et la rediriger sur l’entrée des «Reducer»
• récupérer la sortie standard des «Reducer» et créer sur le HDFS le fichier de sortie

Ce «workflow» pourrait être matérialisé par la suite de commandes shell ci-dessous

cat /var/log/* | map.php > map.log
cat map.log | reduce.php > reduce.log

Vous remarquerez que la vitesse de traitement n’est pas exceptionnel dans le cas de notre démo. Le temps de traitement en ligne de commandes (ci-dessus) prend environ 1 seconde. Le traitement par Hadoop beaucoup plus. Par contre, plus le volume de données de traitement sera important, plus Hadoop répondra à nos besoins. Il est possible d’améliorer la vitesse d’exécution de quelques secondes pour des milliers de lignes de log. Il est préférable que les logs à traiter, soient au sein du HDFS plutôt que sur le file système local. Il est aussi intéressant d’utiliser «Thrift server» beaucoup plus rapide que les commandes natives d’Hadoop pour la manipulation du HDFS.

Il faut savoir, qu’il y aura toujours un laps de temps non négligeable imputé au lancement des jobs de Map / Reduce quelque soit le nombre de lignes à traiter. Ci-dessous, je vous propose le script shell de bootstrap (testé sous Mac os X). Il est perfectible et pourrait être écrit dans un autre langage plus portable comme python.

#!/bin/bash

###conf : à adpater
#Chemin du log ou du rep de logs
PATH_LOG='log'
#Nom du streamer jar : dépend de votre version
STREAM_JAR_NAME="hadoop-streaming-1.0.0.jar"

#Variable d’environnement de travail
APP_PATH=$(readlink -f $(dirname $0))
YEAR=$( date "+%Y" )
MONTH=$( date "+%m" )
DAY=$( date "+%d" )
MIN=$( date "+%M" )
SECOND=$( date "+%S" )
DATE_START=${YEAR}${MONTH}${DAY}
RANDOM_NAME=${DATE_START}${MIN}${SECOND}'-'$RANDOM

#Home d’installation d’hadoop suivant la partie 1 du tuto
HADOOP_HOME_BIN='/opt/hadoop-1.0.0/'

#Création du répertoire de travail sur le HDFS
HDFS_INPUT='input/'$RANDOM_NAME
echo "Création du répertoire de travail : $HDFS_INPUT"
$HADOOP_HOME_BIN/bin/hadoop fs -mkdir $HDFS_INPUT

#On dépose le ou les fichiers de log a traiter
echo "Copie du log $PATH_LOG sur le hdfs $PATH_LOG pour le traitement"
$HADOOP_HOME_BIN/bin/hadoop fs -put $PATH_LOG/* $HDFS_INPUT

#Définition du répertoire où sera entreposé le résultat du Job.
#Attention ! il ne faut pas le créer : Hadoop se chargera de le faire
HDFS_OUTPUT='OUTPUT/'$RANDOM_NAME

#Définition du répertoire de stockage
HDFS_STORAGE='STORAGE/'$DATE_START
echo "Création du répertoire de stockage"
$HADOOP_HOME_BIN/bin/hadoop fs -mkdir $HDFS_STORAGE

#Map Reduce en action
echo "Lancement de MAP / REDUCE"
$HADOOP_HOME_BIN/bin/hadoop jar \
$HADOOP_HOME_BIN/contrib/streaming/$STREAM_JAR_NAME \
-input $HDFS_INPUT \
-output $HDFS_OUTPUT \
-mapper map.php \
-reducer reduce.php \
-file $APP_PATH/map.php \
-file $APP_PATH/reduce.php &
wait %1
STATUS=$?
#Tout c’est bien passé on va bouger le résultat dans notre répertoire de stockage
if [ $STATUS -eq 0 ]
then
    #on copie le résulta dans notre entrepôt et on le renomme
    echo "Copie du résulta $RANDOM_NAME du traitement dans $HDFS_STORAGE"
    $HADOOP_HOME_BIN/bin/hadoop fs -cp $HDFS_OUTPUT/part-* $HDFS_STORAGE/$RANDOM_NAME
    #On sup le répertoire de travail & de sortie
    echo "Supp de l'environnement de travail"
    $HADOOP_HOME_BIN/bin/hadoop fs -rmr $HDFS_INPUT/
    $HADOOP_HOME_BIN/bin/hadoop fs -rmr $HDFS_OUTPUT

    $HADOOP_HOME_BIN/bin/hadoop fs -ls $HDFS_STORAGE
else
    echo "pb avec le Map & Reduce"
fi

N’oubliez pas de modifier les permission du bootstrap afin de le rendre exécutable :

chmod 755 bootstrap.sh

Script PHP

Les scripts de Map et de Reduce ont pour rôle de rendre les données brutes exploitables par d’autres mécanismes d’extraction. Dans notre cas, nous avons un log qui a déjà été qualifié. Le but du Map et du Reduce est d’agréger les données et de les regrouper par [IP - DOMAIN] et de rajouter un compteur d’agrégation.

#!/usr/bin/env php
<?php
## Script du map

$ressources = fopen('php://stdin', 'r');

#on agrège les lignes
$row = array();
while ( ($line = fgets($ressources)) !== false) {
    $data = json_decode($line);

    if(isset($row[$data->ip][$data->domain])) {
        $row[$data->ip][$data->domain]['count'] ++;
    }else {
        $row[$data->ip][$data->domain]['count'] = 1;
    }
}

#on reconstruit le flux de sortie
foreach ($row as $ip=>$value) {
    foreach($value as $keyDomaine => $valueDomain) {
        $line = array('ip'=>$ip,'domain'=>$keyDomaine,'count'=>$valueDomain['count']);
        echo json_encode($line) , PHP_EOL;
    }
}

#!/usr/bin/env php
<?php
## Script du reducer

$ressources = fopen('php://stdin', 'r');

//on check les lignes : et on agrege les lignes
$row = array();
while ( ($line = fgets($ressources)) !== false) {
    $data = json_decode($line);

   if($data->ip == '' && $data->domain == '')
    continue;

    if(isset($row[$data->ip][$data->domain])) {
        $row[$data->ip][$data->domain]['count'] = $row[$data->ip][$data->domain]['count'] + $data->count ;
    }else {
        $row[$data->ip][$data->domain]['count'] = $data->count;
    }

}

//on reconstruit le flux de sortie
foreach ($row as $ip=>$value) {
    foreach($value as $keyDomaine => $valueDomain) {
        echo $ip, chr(9) , $keyDomaine , chr(9), $valueDomain['count'] , PHP_EOL;
    }
}

Exécution MAP & REDUCE

Avant d’exécuter le bootstrap, vous devez passer sous l’identifiant «Hadoop» et avoir démarré «Hadoop».

su hadoop
./bootstrap.sh

#Affichage du résultat stocké sur le HDFS (commande à adapter en fonction de votre test)
hadoop fs -cat STORAGE/20120405/201204050706-11632
124.115.0.16		plop-info.com/	915
124.115.0.16		www.plop-info.com	915
124.115.0.16		www.plop-info.com/	6405
144.85.196.165	www.stephane-raymond.com	6405
157.55.17.146	www.stephane-raymond.com	915
207.46.194.103	www.tec-info.com	2745
207.46.194.53	www.tec-info.com	2745
207.46.204.189	stephane-raymond.com	915
208.115.113.84	plop-info.com	915
66.249.72.136	plop.fr	915
66.249.72.77		www.plop.com	915
80.67.166.230	http://www.plop.com/	915
80.67.166.230	www.integration.com	930
80.67.166.230	www.tec-info	915
90.19.127.168	-	915
90.19.127.168	www.stephane-raymond.info	1830
90.28.60.239	www.mora-mora.fr	12825
91.121.119.198	-	915

Je vous propose de renouveler l’appel au bootstrap.sh. Il va générer plusieurs fois le même contenu avec un nom de fichier «random» dans le HDFS. Cela va nous permettre de mettre en évidence les possibilités offertes par HIVE.

Il est à noter que la comparaison linéaire des fichiers de sortie, en vue de faire des tests unitaires n’est pas possible avec Hadoop. Le «streamer» choisit lui même la répartition des lignes des fichiers de log à envoyer au «Mapper» et aux «Reducer». De ce fait, il est très complexe, voir impossible, de faire de la comparaison de fichiers ligne à ligne, car la distribution est aléatoire. Par contre, ces tests peuvent être fait en ligne de commande en simulant tour à tour le Map et le Reduce.

#Génération du fichier de sorti du Map qui pourra être comparé avec votre fichier de test
cat monfichier.log | map.php > map.txt

#Génération du fichier de sorti du Reduce
cat map.txt | reduce.php > reduce.txt

#on peut comparer map.txt et reduce.txt à notre étalon de test

Nous avons jusqu’ici travaillé sur la structuration et le stockage de la data (l’entrepôt de données). Voyons maintenant l’étage «warehouse», ou comment extraire la donnée pour le «reporting».

HIVE outil d’extraction Open Source

HIVE est une solution de «DataWareHouse» construit au dessus d’«Hadoop». Au delà de son langage de type «SQL» (HiveQL), il s’appuie sur le traitement distribué d’Hadoop et l’implémentation de Map Reduce. Toutefois, il ne faut pas voir «HIVE» comme une solution permettant d’extraire de manière instantanée des données. Les applications web ayant besoin d’instantanéité, il sera préférable d’adjoindre à l’architecture technique un SGBD adapté à la restitution rapide de l’information : Cassandra ; Redis ; MongoDB ; MySQL et pourquoi un mix (à voir en fonction de vos besoins).

Nous allons présenter, installer et utiliser HIVE à travers une de ses particularités qui permet de faire un «mapping» entre la donnée stockée sur le HDFS et la définition d’un schéma ou table au sens MySQL par exemple. Nous allons définir la structure de la table, et indiquer à HIVE, la correspondance de cette dernière sur le HDFS.

• il restera à écrire les requêtes HQL d’extractions,
• récupérer le résultat dans un format approprié,
• traiter éventuellement cette extraction pour une intégration sur un SGBD propice aux applications WEB et donc au reporting graphique.

Le volume des données manipulées étant massive, elle s’accompagne dans la plus part des cas d’une dimension temporelle. C’est pourquoi, notre exemple s’appuiera sur un mécanisme de partitionnement.

Installation et paramétrage de HIVE

HIVE s’installe de la même manière et aussi simplement qu’«Hadoop». Il sera installé en mode «mono session» : installation par défaut. Pour faire simple, HIVE a besoin de stocker des informations «metadata» pour garder en mémoire les «mapping» et les structures de tables qui auront été définies. Pour cela, HIVE utilise un «metastrore» local utilisable via «derby». C’est pour cette raison que HIVE n’est utilisable que dans une version mono session. Il est possible de configurer HIVE avec MySQL afin de pouvoir lancer plusieurs sessions en parallèle .

#préparation du répertoire d’installation
mkdir /opt
cd /opt/

#téléchargez le binaire ou les sources version stable
wget http://mirror.speednetwork.de/apache/hive/stable/hive-0.8.1-bin.tar.gz

#décompression & préparation
tar -xvf hive-0.8.1-bin.tar.gz
mv hive-0.8.1-bin hive-0.8.1
chown -R hadoop:hadoop /opt/hive-0.8.1
ln -s hive-0.8.1 hive

#accès en mode cli et utilisation
cd hive
bin/hive

Notre outil d’extraction en place, il nous reste plus qu’à définir notre table et faire le «mapping» avec les données présentes sur le HDFS. Pour rappel, dans la première partie du billet, nous avons exécuté un job de Map Reduce et stocké le résultat au sein du HDFS :

hadoop fs -ls STORAGE/20120405
Found 2 items
-rw-r--r--   1 hadoop supergroup        624 2012-04-05 08:08 /user/hadoop/STORAGE/20120405/201204050706-11632
-rw-r--r--   1 hadoop supergroup        624 2012-04-05 08:11 /user/hadoop/STORAGE/20120405/201204050955-11903

Le «mapping» va donc consister à lier le contenu du répertoire avec notre table et de définir un partitionnement sur la date. Il est possible de faire des requêtes HIVE avec des données au format JSON. Pour cela, il faut utiliser une librairie connexe : cela fera l’objet certainement d’un autre article.

#première solution : en mode cli
cd /opt/hive-0.8.1/
bin/hive
CREATE EXTERNAL TABLE IF NOT EXISTS ip_domain (ip STRING, domain STRING,count INT) PARTITIONED BY (day INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE LOCATION '/user/hadoop/STORAGE/';
show tables;
desc ip_domain;

#deuxième solution : peut être utilisé au sein d’un script shell par exemple
drop table ip_domain;
exit;

bin/hive -e "CREATE EXTERNAL TABLE IF NOT EXISTS ip_domain (ip STRING, domain STRING,count INT) PARTITIONED BY (day INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE LOCATION '/user/hadoop/STORAGE/' "

bin/hive
show tables;
exit;

#vérifier bien à être dans le répertoire de HIVE !
cd /opt/hive-0.8.1
#création de la partition et link sur le répertoire du HDFS
bin/hive -e "ALTER TABLE ip_domain ADD PARTITION (day='20120405') LOCATION '/user/hadoop/STORAGE/20120405/';"

#Pour supprimer une partition :
bin/hive -e "ALTER TABLE ip_domain DROP PARTITION (day='20120405');"

Faisons nos premières requêtes HIVE. Le HQL est très proche du SQL voir identique. Toutefois, certaines fonctions ne sont pas encore disponibles comme les requêtes imbriquées (dans certaines conditions). Il est possible de faire des requêtes avec des «jointures» ou opérer des fonctions mathématiques. Au lancement de votre requête, HIVE va analyser la structure sémantique de la requête et va ensuite distribuer le traitement et utiliser Map / Reduce comme le montre l’image ci-dessous.

#extraction de data et affichage dans la console
cd /opt/hive-0.8.1
SELECT count(*) FROM ip_domain;
SELECT domain,count(1) as nb_ip,sum(count) as hits FROM ip_domain WHERE day=20120405 GROUP BY domain;

#extraction de data et intégration du résultat dans un fichier txt
bin/hive -e " SELECT domain,count(1) as nb_ip,sum(count) as hits FROM ip_domain WHERE day=20120405 GROUP BY domain; " > monfichier.txt
#Les éléments des lignes sont séparés par une tabulation !

HIVE un outil d’extraction puissant

HADOOP et HIVE est un duo très intéressant pour manipuler et stocker de grosse volumétrie de données. La donnée reste le maillon faible de la chaîne. Il est primordial que cette dernière soit parfaitement structurée et nécessite donc un traitement de fond. N’ayant pas encore assez de recul sur une utilisation poussée d’Hadoop, je préconise le couplage avec un SGBD qui permettra de stocker les data issue des requêtes HIVE. Cela permet entre autre, d’extraire de manière rapide les informations requises pour les tableaux de reporting Marketing.