<jupyter_start><jupyter_text>Map/Reduce avec PIG sur Azure - énoncéManipulation de fichiers sur un cluster Hadoop [HDInsight](https://azure.microsoft.com/en-us/services/hdinsight/), premieer job map/reduce avec [PIG](https://pig.apache.org/). Lire aussi [From Pig to Spark: An Easy Journey to Spark for Apache Pig Developers](https://mapr.com/blog/easy-journey-from-pig-to-spark-for-developers/).<jupyter_code>from jyquickhelper import add_notebook_menu
add_notebook_menu()<jupyter_output><empty_output><jupyter_text>Dans le cadre du [partenariat avec Microsoft](http://www.microsoft.com/france/Hub-Presse/communiques-de-presse/fiche-communique.aspx?eid=f7e7f695-fb08-4c6d-b4ec-3cde562ba429), l'ENSAE dispose de ressources Azure. Azure est une plateforme qui propose de nombreux [fonctionnalités et services](http://azure.microsoft.com/fr-fr/services/cloud-services/). Les élèves n'auront pas directement accès au portail Azure mais recevront les identifiants et mots de passe pour les services mis en place à leur intention. Ils seront différents selon les services mais ce sera le même identifiant pour chaque élève.Cette séance de travaux pratique s'appuiera sur le jeu de données suivant : [Localization Data for Person Activity Data Set](https://archive.ics.uci.edu/ml/datasets/Localization+Data+for+Person+Activity) qu'on récupère comme suit :<jupyter_code>import pyensae.datasource
import urllib.error
try:
    # ce soir ça ne marche pas...
    pyensae.datasource.download_data("ConfLongDemo_JSI.txt", 
                          website="https://archive.ics.uci.edu/ml/machine-learning-databases/00196/")
except urllib.error.URLError:
    # donc je récupère une vieille copie
    pyensae.datasource.download_data("ConfLongDemo_JSI.zip")<jupyter_output><empty_output><jupyter_text>On l'insère dans une base de données SQL.<jupyter_code>columns = "sequence tag timestamp dateformat x y z activity".split()
import pandas, sqlite3, os
df = pandas.read_csv("ConfLongDemo_JSI.txt", sep=",", names=columns)
if os.path.exists("ConfLongDemo_JSI.db3"):
    os.remove("ConfLongDemo_JSI.db3")
con = sqlite3.connect("ConfLongDemo_JSI.db3")
df.to_sql("person", con)
con.close()
df.head()<jupyter_output><empty_output><jupyter_text>On crée un petit exemple pour plus tard qu'on enregistre également dans la base de données.<jupyter_code>columns = "sequence tag timestamp dateformat x y z activity".split()
import pandas
df = pandas.read_csv("ConfLongDemo_JSI.txt", sep=",", names=columns)
dfs = df[:1000]
dfs.to_csv("ConfLongDemo_JSI.small.txt", header=False)

import sqlite3
con = sqlite3.connect("ConfLongDemo_JSI.db3")
dfs.to_sql("person_small", con)
con.close()<jupyter_output><empty_output><jupyter_text>Partie 0 : aperçu de Azure HDInsight    Il faut se connecter au cluster avec l'url : ``https://....datascience.fr/....&login=``.<jupyter_code>from pyquickhelper.helpgen import NbImage
NbImage("clhdins.png", width=600)<jupyter_output><empty_output><jupyter_text>On peut accéder à ce cluster via une adresse web qui s'ouvre sur la fenêtre suivante d'où on peut voir le contenu des fichiers et<jupyter_code>NbImage("clhdins2.png", width=600)<jupyter_output><empty_output><jupyter_text>Le fenêtre suivante permet d'effectuer des requêtes distribuées avec le langage [Hive](https://hive.apache.org/) très proche du SQL.<jupyter_code>NbImage("clhdins3.png", width=600)<jupyter_output><empty_output><jupyter_text>Partie 1 :  manipulation de fichiersAvant de commencer à déplacer des fichiers, il faut comprendre qu'il y a trois entités :* l'ordinateur local (celui dont vous vous servez)* le stockage : [Azure Blob Storage](http://azure.microsoft.com/en-us/documentation/articles/storage-dotnet-how-to-use-blobs/)* le cluster  [Hadoop](http://fr.hortonworks.com/)<jupyter_code>from IPython.core.display import Image
NbImage("azure3.png")<jupyter_output><empty_output><jupyter_text>Pour exécuter une tâche Map/Reduce, il faut suivre le processus suivant :* 1a : on communique avec l'Azure Blob Storage pour **uploader** les données avec lesquelles on va travailler* 1b : on **uploade** également la description de la tâche (ici un script PIG)* 2 : on envoie une requête au cluster pour demander l'exécution d'une tâche* 3 : le cluster exécute la tâche, il échangera des données avec le storage* 4 : on **downloade** les résultats sur son ordinateur localPour manipuler les données sur un cluster, il faut d'abord les [uploader](http://fr.wiktionary.org/wiki/uploader) sur ce cluster. Pour les récupérer, il faut les [downloader](http://fr.wikipedia.org/wiki/T%C3%A9l%C3%A9chargement). Pour faciliter les choses, on va utiliser des [commandes magiques](http://ipython.org/ipython-doc/dev/interactive/magics.html?highlight=command) implémentées dans le module [pyensae](http://www.xavierdupre.fr/app/pyensae/helpsphinx/) (>= 0.9). La première tâche est d'enregistrer dans l'espace de travail le nom du blob storage, l'alias et le mot de passe dans le workspace du notebook. Ces commandes utilise le module [azure](https://github.com/Azure/azure-sdk-for-python) qu'il faut installer.Le code est avant tout destiné à **ne pas laisser votre mot de passe en clair dans le notebook**. S'il est en clair, tôt ou tard, vous oublierez de l'effacer avant de partager votre notebook. Dans ce cas, il faut changer de mot de passe sans tarder.<jupyter_code>from pyquickhelper.ipythonhelper import open_html_form
params={"blob_storage":"", "password":""}
open_html_form(params=params,title="server + credentials", key_save="blob")
blobstorage = blob["blob_storage"]
blobpassword = blob["password"]<jupyter_output><empty_output><jupyter_text>On ouvre la connection au blob storage.<jupyter_code>cl, bs = %blob_open
cl, bs<jupyter_output><empty_output><jupyter_text>On regarde la liste des blob storage disponibles.<jupyter_code>l = %blob_containers
l<jupyter_output><empty_output><jupyter_text>On regarde les listes des fichiers dans un répertoire du blob storage :<jupyter_code>df = %blob_ls clusterensaeazure1
df.head()<jupyter_output><empty_output><jupyter_text>Et plus particulièrement :<jupyter_code>df=%blob_ls clusterensaeazure1/testensae
df<jupyter_output><empty_output><jupyter_text>On supprime tous les fichiers de répertoire :<jupyter_code>for name in df["name"]:
    cl.delete_blob(bs, "clusterensaeazure1", name)<jupyter_output><empty_output><jupyter_text>On vérifie qu'il ne reste plus rien :<jupyter_code>%blob_ls clusterensaeazure1/testensae<jupyter_output><empty_output><jupyter_text>On uploade un fichier dans le blob storage :<jupyter_code>%blob_up ConfLongDemo_JSI.small.txt clusterensaeazure1/testensae/ConfLongDemo_JSI.small.txt<jupyter_output><empty_output><jupyter_text>On vérifie qu'il a bien été uploadé :<jupyter_code>%blob_ls clusterensaeazure1/testensae<jupyter_output><empty_output><jupyter_text>On récupère le fichier uploadé :<jupyter_code>if os.path.exists("ConfLongDemo_JSI.small2.txt") : os.remove("ConfLongDemo_JSI.small2.txt")
%blob_down clusterensaeazure1/testensae/ConfLongDemo_JSI.small.txt ConfLongDemo_JSI.small2.txt<jupyter_output><empty_output><jupyter_text>On vérifie que le fichier a été téléchargé :<jupyter_code>%lsr -f Conf.*[.]txt<jupyter_output><empty_output><jupyter_text>On peut également faire une copie d'un blob :<jupyter_code>%blob_copy clusterensaeazure1/testensae/ConfLongDemo_JSI.small.txt clusterensaeazure1/testensae/ConfLongDemo_JSI.small_copy.txt<jupyter_output><empty_output><jupyter_text>Et supprimer le blob copié :<jupyter_code>%blob_delete clusterensaeazure1/testensae/ConfLongDemo_JSI.small_copy.txt<jupyter_output><empty_output><jupyter_text>On termine la session.<jupyter_code>%blob_close<jupyter_output><empty_output><jupyter_text>Partie 2 : premier job map/reduce avec PIGPour cette partie, l'idée d'exécuter des jobs Map/Reduce sur le fichier ``ConfLongDemo_JSI.small.example.txt`` puis de vérifier qu'on obtient bien les même résultats sur le même fichier en utilisant une requête SQL. Le code pour créer la connexion à la passerelle est recopié ci-dessous mais il n'est pas nécessaire de l'exécuter si la connexion n'a pas été interrompue.<jupyter_code>from pyquickhelper.ipythonhelper import open_html_form
params={"blob_storage":"", "password1":"", "hadoop_server":"", "password2":"", "username":"xavierdupre"}
open_html_form(params=params,title="server + hadoop + credentials", key_save="blobhp")
import pyensae
%load_ext pyensae
%load_ext pyenbc
blobstorage = blobhp["blob_storage"]
blobpassword = blobhp["password1"]
hadoop_server = blobhp["hadoop_server"]
hadoop_password = blobhp["password2"]
username = blobhp["username"]
client, bs = %hd_open
client, bs<jupyter_output><empty_output><jupyter_text>Le cluster est lié à un autre container ``hdblobstorage`` (différent de celui utiliser pour la première partie. Il faut uploader les données à cet endroit. On supprime d'abord les fichiers existant :<jupyter_code>df = %blob_ls hdblobstorage/testensae/
for name in df["name"]:
    client.delete_blob(bs, "hdblobstorage", name)
%blob_up ConfLongDemo_JSI.small.txt hdblobstorage/testensae/ConfLongDemo_JSI.small.txt
%blob_ls hdblobstorage/testensae<jupyter_output><empty_output><jupyter_text>JobsUn job définit l'ensemble des traitements que vous souhaitez effectuer sur un ou plusieurs fichiers. Le langage [PIG](http://pig.apache.org/docs/r0.12.1/basic.html) permet de décrire ces traitements. Le programme est ensuite interprété puis soumis à Hadoop qui s'occupe de répartir de traitements sur l'ensemble des resources dont il dispose. La commande suivante permet d'obtenir l'ensemble des tâches associés aux jobs :  SELECT ... WHERE<jupyter_code>import sqlite3
con = sqlite3.connect("ConfLongDemo_JSI.db3")

import pandas.io.sql as psql
sql = 'SELECT * FROM person_small WHERE activity == "walking"'
df = psql.read_sql(sql, con)
con.close()
df.tail()<jupyter_output><empty_output><jupyter_text>Avec [PIG](http://en.wikipedia.org/wiki/Pig_Latin) ([syntaxe](http://pig.apache.org/docs/r0.12.0/basic.html)), le programme inclut trois étapes :* la déclaration de l'entrée (le fichier ``ConfLongDemo_JSI.small.example.txt``) (voir [LOAD](https://pig.apache.org/docs/r0.9.1/basic.htmlload))* la tâche à proprement parler (voir [FILTER](https://pig.apache.org/docs/r0.9.1/basic.htmlfilter))* la création de la sortie (le fichier ``ConfLongDemo_JSI.small.example.walking.txt``) (voir [STORE](https://pig.apache.org/docs/r0.9.1/basic.htmlstore))La commande magique permet d'écrire un fichier ``select1.pig`` avec l'encoding ``utf-8`` (les caractères accentuées sont possibles).<jupyter_code>%%PIG_azure select_azure_2015.pig

myinput = LOAD '$CONTAINER/testensae/ConfLongDemo_JSI.small.txt' 
          using PigStorage(',') 
          AS (index:long, sequence, tag, timestamp:long, dateformat, x:double,y:double, z:double, activity) ;

filt = FILTER myinput BY activity == 'walking' ;

STORE filt INTO '$CONTAINER/$PSEUDO/testensae/ConfLongDemo_JSI.small.walking_2015.txt' USING PigStorage() ;<jupyter_output><empty_output><jupyter_text>La commande suivante fait deux choses : elle uploade le job sur le blob storage et soumet une requête au cluster afin de l'exécuter :<jupyter_code>jid = %hd_pig_submit select_azure_2015.pig
jid<jupyter_output><empty_output><jupyter_text>Après quelques essais infructueux, on s'aperçoit parfois que les numéros de job vont de deux en deux (sauf si le job ne compile pas). Le protocole utilisé pour communiquer avec Azure HD Insight est [WebHCat](https://cwiki.apache.org/confluence/display/Hive/WebHCat+Reference). Lorsque le service reçoit l'instruction, il lance un autre job et façon à soumettre le job au cluster. WebHCat sert d'intermédiaire.On vérifie que le script a bien été uploadé :<jupyter_code>df = %blob_ls /$PSEUDO/scripts/pig/
df[df.name.apply(lambda x : "select" in x)]<jupyter_output><empty_output><jupyter_text>On exécute la commande suivante pour connaître le status de l'exécution. On l'execute à nouveau de temps en temps pour vérifier que le job s'est bien déroulé.<jupyter_code>st = %hd_job_status jid["id"]
st["id"],st["percentComplete"],st["completed"],st["status"]["jobComplete"],st["status"]["state"]<jupyter_output><empty_output><jupyter_text>On vérifie le contenu du cluster :<jupyter_code>df=%blob_ls /$PSEUDO/testensae
df[df.last_modified.apply(lambda x : "2015" in x)]
set(df[df.last_modified.apply(lambda x : "2015" in x)]["name"])<jupyter_output><empty_output><jupyter_text>Le job s'est bien exécuté. On downloade le contenu sur l'ordinateur :<jupyter_code>import os
if os.path.exists("results.txt"):os.remove("results.txt")
%blob_down /$PSEUDO/testensae/ConfLongDemo_JSI.small.keep_walking.txt/part-m-00000 results.txt<jupyter_output><empty_output><jupyter_text>La plupart du temps, il y aura plus d'un fichier en sortie. C'est pour cela qu'on utilise la commande magique ``%blob_downmerge`` qui va télécharger tous les fichiers et les regrouper localement en un seul :<jupyter_code>if os.path.exists("results_allfiles.txt"):os.remove("results_allfiles.txt")
%blob_downmerge /$PSEUDO/testensae/ConfLongDemo_JSI.small.keep_walking.txt results_allfiles.txt
%lsr r.*s[.]txt<jupyter_output><empty_output><jupyter_text>On regarde les logs de l'exécution. Pour le moment, les logs sont stockés dans le même fichier quelque soit le job.<jupyter_code>if os.path.exists("stderr.txt"): os.remove("stderr.txt")
%blob_ls /$SCRIPTPIG/select_azure.pig.log
df = %blob_ls /$SCRIPTPIG/select_azure_2015.pig.log
list(df["name"])<jupyter_output><empty_output><jupyter_text>L'instruction récupère l'erreur standard pour un job donné :<jupyter_code>%hd_tail_stderr jid["id"]<jupyter_output><empty_output><jupyter_text>Et pour tuer un job, il sufft d'appeler :<jupyter_code>%hd_job_kill <job_id><jupyter_output><empty_output><jupyter_text>Partie 3 : syntaxe PIG et exercicesDans cette partie, l'objectif est de transcrire un ``GROUP BY`` en PIG, un ``JOIN`` et de combiner toutes ces opérations en un seul job au cours du second exercice. Ces exemples utilisent de petits fichiers. Utiliser un job Map/Reduce n'a pas beaucoup d'intérêt à moins que la taille de ces fichiers n'atteigne un giga-octets. Les instructions sont à chercher dans cette page : [Pig Latin Basics](http://pig.apache.org/docs/r0.12.0/basic.html).  Exercice 1 : GROUP BY<jupyter_code>import pandas, sqlite3
con = sqlite3.connect("ConfLongDemo_JSI.db3")
df = pandas.read_sql("""SELECT activity, count(*) as nb FROM person GROUP BY activity""", con)
con.close()
df.head()<jupyter_output><empty_output><jupyter_text>Il faut maintenant le faire avec PIG.  Exercice 2 : JOIN<jupyter_code>con = sqlite3.connect("ConfLongDemo_JSI.db3")
df = pandas.read_sql("""SELECT person.*, A.nb FROM person INNER JOIN (
                            SELECT activity, count(*) as nb FROM person GROUP BY activity) AS A
                            ON person.activity == A.activity""", con)
con.close()
df.head()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11028275908388743, 'fraction_numerical': 0.009402914903620122, 'mean_word_length': 6.505040322580645, 'pattern_counts': {'":': 7, '<': 150, '<?xml version=': 0, '>': 151, 'https://': 12, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '37', 'dir': 'jupyter-structured-clean-dedup', 'id': '112150', 'license': 'MIT', 'n_tokens_mistral': 5399, 'n_tokens_neox': 5289, 'n_words': 1296, 'path': '_doc/notebooks/pig_hive/pig_azure.ipynb', 'repo_name': 'sdpython/sparkouille', 'size': '14889'}