.. _pigazurerst:

======================================
Map/Reduce avec PIG sur Azure - énoncé
======================================


.. only:: html

    **Links:** :download:`notebook <pig_azure.ipynb>`, :downloadlink:`html <pig_azure2html.html>`, :download:`PDF <pig_azure.pdf>`, :download:`python <pig_azure.py>`, :downloadlink:`slides <pig_azure.slides.html>`, :githublink:`GitHub|_doc/notebooks/pig_hive/pig_azure.ipynb|*`


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/>`__.

.. code:: ipython3

    from jyquickhelper import add_notebook_menu
    add_notebook_menu()






.. contents::
    :local:





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 :

.. code:: ipython3

    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")

On l’insère dans une base de données SQL.

.. code:: ipython3

    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()






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>sequence</th>
          <th>tag</th>
          <th>timestamp</th>
          <th>dateformat</th>
          <th>x</th>
          <th>y</th>
          <th>z</th>
          <th>activity</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>A01</td>
          <td>010-000-024-033</td>
          <td>633790226051280329</td>
          <td>27.05.2009 14:03:25:127</td>
          <td>4.062931</td>
          <td>1.892434</td>
          <td>0.507425</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>1</th>
          <td>A01</td>
          <td>020-000-033-111</td>
          <td>633790226051820913</td>
          <td>27.05.2009 14:03:25:183</td>
          <td>4.291954</td>
          <td>1.781140</td>
          <td>1.344495</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>2</th>
          <td>A01</td>
          <td>020-000-032-221</td>
          <td>633790226052091205</td>
          <td>27.05.2009 14:03:25:210</td>
          <td>4.359101</td>
          <td>1.826456</td>
          <td>0.968821</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>3</th>
          <td>A01</td>
          <td>010-000-024-033</td>
          <td>633790226052361498</td>
          <td>27.05.2009 14:03:25:237</td>
          <td>4.087835</td>
          <td>1.879999</td>
          <td>0.466983</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>4</th>
          <td>A01</td>
          <td>010-000-030-096</td>
          <td>633790226052631792</td>
          <td>27.05.2009 14:03:25:263</td>
          <td>4.324462</td>
          <td>2.072460</td>
          <td>0.488065</td>
          <td>walking</td>
        </tr>
      </tbody>
    </table>
    </div>



On crée un petit exemple pour plus tard qu’on enregistre également dans
la base de données.

.. code:: ipython3

    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()

Partie 0 : aperçu de Azure HDInsight
------------------------------------

Il faut se connecter au cluster avec l’url :
``https://....datascience.fr/....&login=<login>``.

.. code:: ipython3

    from pyquickhelper.helpgen import NbImage
    NbImage("clhdins.png", width=600)




.. image:: pig_azure_9_0.png
   :width: 600px



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

.. code:: ipython3

    NbImage("clhdins2.png", width=600)




.. image:: pig_azure_11_0.png
   :width: 600px



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.

.. code:: ipython3

    NbImage("clhdins3.png", width=600)




.. image:: pig_azure_13_0.png
   :width: 600px



Partie 1 : manipulation de fichiers
-----------------------------------

Avant 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/>`__

.. code:: ipython3

    from IPython.core.display import Image
    NbImage("azure3.png")




.. image:: pig_azure_15_0.png



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 local

Pour 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.

.. code:: ipython3

    from pyquickhelper.ipythonhelper import open_html_form
    params={"blob_storage":"", "password":""}
    open_html_form(params=params,title="server + credentials", key_save="blob")






.. raw:: html

    <div style="background-color:gainsboro; padding:2px; border:0px;"><b>server + credentials</b>
    <br />blob_storage <input type="text" id="blobblob_storage" value="" size="80" />
    <br />password <input type="password" id="blobpassword" value="" size="80" />
    <br /><button onclick="set_valueblob()">Ok</button></div>
    <script type="text/Javascript">
    function blobcallback(msg) {
       var ret = msg.content.data['text/plain'];
       $('#outblob').text(ret);
    }
    function set_valueblob(){
       command='blob = {' ;
       var blobblob_storagevar_value = document.getElementById('blobblob_storage').value;
       command += '"blob_storage":"' + blobblob_storagevar_value + '",';
       var blobpasswordvar_value = document.getElementById('blobpassword').value;
       command += '"password":"' + blobpasswordvar_value + '",';
       command += '}';
       var kernel = IPython.notebook.kernel;
       kernel.execute(command);
    }
    </script>



.. code:: ipython3

    blobstorage = blob["blob_storage"]
    blobpassword = blob["password"]

On ouvre la connection au blob storage.

.. code:: ipython3

    cl, bs = %blob_open
    cl, bs




.. parsed-literal::
    (<pyensae.remote.azure_connection.AzureClient at 0xaa63f60>,
     <azure.storage.blob.blobservice.BlobService at 0xb46c908>)



On regarde la liste des blob storage disponibles.

.. code:: ipython3

    l = %blob_containers
    l




.. parsed-literal::
    ['clusterensaeazure1',
     'clusterensaeazure1-1',
     'clusterensaeazure2',
     'clusterensaeazure2-1',
     'hdblobstorage',
     'petittest',
     'sparkclus',
     'sparkclus2',
     'testhadoopensae']



On regarde les listes des fichiers dans un répertoire du blob storage :

.. code:: ipython3

    df = %blob_ls clusterensaeazure1
    df.head()






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>last_modified</th>
          <th>content_type</th>
          <th>content_length</th>
          <th>blob_type</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>DonnotLaugel/mv_0000001.txt</td>
          <td>Sun, 01 Feb 2015 11:10:47 GMT</td>
          <td>application/octet-stream</td>
          <td>11233</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>1</th>
          <td>HdiSamples/SensorSampleData/building/building.csv</td>
          <td>Sun, 16 Nov 2014 21:24:42 GMT</td>
          <td>application/octet-stream</td>
          <td>544</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>2</th>
          <td>HdiSamples/SensorSampleData/hvac/HVAC.csv</td>
          <td>Sun, 16 Nov 2014 21:24:42 GMT</td>
          <td>application/octet-stream</td>
          <td>240591</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>3</th>
          <td>HdiSamples/StorageAnalytics/hive-serde-microso...</td>
          <td>Sun, 16 Nov 2014 21:24:42 GMT</td>
          <td>application/octet-stream</td>
          <td>9562</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>4</th>
          <td>HdiSamples/StorageAnalytics/hive-serde-microso...</td>
          <td>Sun, 16 Nov 2014 21:24:42 GMT</td>
          <td>application/octet-stream</td>
          <td>10290</td>
          <td>BlockBlob</td>
        </tr>
      </tbody>
    </table>
    </div>



Et plus particulièrement :

.. code:: ipython3

    df=%blob_ls clusterensaeazure1/testensae
    df






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>last_modified</th>
          <th>content_type</th>
          <th>content_length</th>
          <th>blob_type</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>testensae/CASP.csv</td>
          <td>Thu, 05 Feb 2015 13:40:12 GMT</td>
          <td>application/octet-stream</td>
          <td>3528710</td>
          <td>BlockBlob</td>
        </tr>
      </tbody>
    </table>
    </div>



On supprime tous les fichiers de répertoire :

.. code:: ipython3

    for name in df["name"]:
        cl.delete_blob(bs, "clusterensaeazure1", name)

On vérifie qu’il ne reste plus rien :

.. code:: ipython3

    %blob_ls clusterensaeazure1/testensae






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>url</th>
        </tr>
      </thead>
      <tbody>
      </tbody>
    </table>
    </div>



On uploade un fichier dans le blob storage :

.. code:: ipython3

    %blob_up ConfLongDemo_JSI.small.txt clusterensaeazure1/testensae/ConfLongDemo_JSI.small.txt




.. parsed-literal::
    'testensae/ConfLongDemo_JSI.small.txt'



On vérifie qu’il a bien été uploadé :

.. code:: ipython3

    %blob_ls clusterensaeazure1/testensae






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>last_modified</th>
          <th>content_type</th>
          <th>content_length</th>
          <th>blob_type</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>testensae/ConfLongDemo_JSI.small.txt</td>
          <td>Thu, 29 Oct 2015 00:18:55 GMT</td>
          <td>application/octet-stream</td>
          <td>132727</td>
          <td>BlockBlob</td>
        </tr>
      </tbody>
    </table>
    </div>



On récupère le fichier uploadé :

.. code:: ipython3

    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




.. parsed-literal::
    'ConfLongDemo_JSI.small2.txt'



On vérifie que le fichier a été téléchargé :

.. code:: ipython3

    %lsr -f Conf.*[.]txt






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>directory</th>
          <th>last_modified</th>
          <th>name</th>
          <th>size</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>False</td>
          <td>2015-10-29 00:39:34.239955</td>
          <td>.\ConfLongDemo_JSI.small.example.is_back_local...</td>
          <td>129.62 Kb</td>
        </tr>
        <tr>
          <th>1</th>
          <td>False</td>
          <td>2015-10-29 01:16:41.403741</td>
          <td>.\ConfLongDemo_JSI.small.txt</td>
          <td>129.62 Kb</td>
        </tr>
        <tr>
          <th>2</th>
          <td>False</td>
          <td>2015-10-29 01:19:03.751954</td>
          <td>.\ConfLongDemo_JSI.small2.txt</td>
          <td>129.62 Kb</td>
        </tr>
        <tr>
          <th>3</th>
          <td>False</td>
          <td>2015-10-29 00:25:23.606115</td>
          <td>.\ConfLongDemo_JSI.txt</td>
          <td>20.55 Mb</td>
        </tr>
      </tbody>
    </table>
    </div>



On peut également faire une copie d’un blob :

.. code:: ipython3

    %blob_copy clusterensaeazure1/testensae/ConfLongDemo_JSI.small.txt clusterensaeazure1/testensae/ConfLongDemo_JSI.small_copy.txt




.. parsed-literal::
    True



Et supprimer le blob copié :

.. code:: ipython3

    %blob_delete clusterensaeazure1/testensae/ConfLongDemo_JSI.small_copy.txt




.. parsed-literal::
    True



On termine la session.

.. code:: ipython3

    %blob_close




.. parsed-literal::
    True



Partie 2 : premier job map/reduce avec PIG
------------------------------------------

Pour 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.

.. code:: ipython3

    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")






.. raw:: html

    <div style="background-color:gainsboro; padding:2px; border:0px;"><b>server + hadoop + credentials</b>
    <br />blob_storage <input type="text" id="blobhpblob_storage" value="" size="80" />
    <br />hadoop_server <input type="text" id="blobhphadoop_server" value="" size="80" />
    <br />password1 <input type="password" id="blobhppassword1" value="" size="80" />
    <br />password2 <input type="password" id="blobhppassword2" value="" size="80" />
    <br />username <input type="text" id="blobhpusername" value="xavierdupre" size="80" />
    <br /><button onclick="set_valueblobhp()">Ok</button></div>
    <script type="text/Javascript">
    function blobhpcallback(msg) {
       var ret = msg.content.data['text/plain'];
       $('#outblobhp').text(ret);
    }
    function set_valueblobhp(){
       command='blobhp = {' ;
       var blobhpblob_storagevar_value = document.getElementById('blobhpblob_storage').value;
       command += '"blob_storage":"' + blobhpblob_storagevar_value + '",';
       var blobhphadoop_servervar_value = document.getElementById('blobhphadoop_server').value;
       command += '"hadoop_server":"' + blobhphadoop_servervar_value + '",';
       var blobhppassword1var_value = document.getElementById('blobhppassword1').value;
       command += '"password1":"' + blobhppassword1var_value + '",';
       var blobhppassword2var_value = document.getElementById('blobhppassword2').value;
       command += '"password2":"' + blobhppassword2var_value + '",';
       var blobhpusernamevar_value = document.getElementById('blobhpusername').value;
       command += '"username":"' + blobhpusernamevar_value + '",';
       command += '}';
       var kernel = IPython.notebook.kernel;
       kernel.execute(command);
    }
    </script>



.. code:: ipython3

    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"]

.. code:: ipython3

    client, bs = %hd_open
    client, bs




.. parsed-literal::
    (<pyensae.remote.azure_connection.AzureClient at 0xa293b00>,
     <azure.storage.blob.blobservice.BlobService at 0xa293ba8>)



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 :

.. code:: ipython3

    df = %blob_ls hdblobstorage/testensae/
    for name in df["name"]:
        client.delete_blob(bs, "hdblobstorage", name)

.. code:: ipython3

    %blob_up ConfLongDemo_JSI.small.txt hdblobstorage/testensae/ConfLongDemo_JSI.small.txt




.. parsed-literal::
    'testensae/ConfLongDemo_JSI.small.txt'



.. code:: ipython3

    %blob_ls hdblobstorage/testensae






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>last_modified</th>
          <th>content_type</th>
          <th>content_length</th>
          <th>blob_type</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>testensae/ConfLongDemo_JSI.small.txt</td>
          <td>Thu, 29 Oct 2015 00:23:00 GMT</td>
          <td>application/octet-stream</td>
          <td>132727</td>
          <td>BlockBlob</td>
        </tr>
      </tbody>
    </table>
    </div>



Jobs
----

Un 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
~~~~~~~~~~~~~~

.. code:: ipython3

    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()






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>index</th>
          <th>sequence</th>
          <th>tag</th>
          <th>timestamp</th>
          <th>dateformat</th>
          <th>x</th>
          <th>y</th>
          <th>z</th>
          <th>activity</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>165</th>
          <td>698</td>
          <td>A01</td>
          <td>010-000-030-096</td>
          <td>633790226277429870</td>
          <td>27.05.2009 14:03:47:743</td>
          <td>2.965434</td>
          <td>1.782434</td>
          <td>0.228563</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>166</th>
          <td>699</td>
          <td>A01</td>
          <td>020-000-032-221</td>
          <td>633790226277970452</td>
          <td>27.05.2009 14:03:47:797</td>
          <td>4.322405</td>
          <td>1.571452</td>
          <td>1.400499</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>167</th>
          <td>700</td>
          <td>A01</td>
          <td>010-000-024-033</td>
          <td>633790226278240749</td>
          <td>27.05.2009 14:03:47:823</td>
          <td>3.133065</td>
          <td>1.769329</td>
          <td>-0.022590</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>168</th>
          <td>701</td>
          <td>A01</td>
          <td>020-000-033-111</td>
          <td>633790226278781331</td>
          <td>27.05.2009 14:03:47:877</td>
          <td>3.121254</td>
          <td>1.549842</td>
          <td>1.048139</td>
          <td>walking</td>
        </tr>
        <tr>
          <th>169</th>
          <td>702</td>
          <td>A01</td>
          <td>020-000-032-221</td>
          <td>633790226279051629</td>
          <td>27.05.2009 14:03:47:907</td>
          <td>3.281498</td>
          <td>1.498734</td>
          <td>0.620412</td>
          <td>walking</td>
        </tr>
      </tbody>
    </table>
    </div>



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.html#load>`__)
-  la tâche à proprement parler (voir
   `FILTER <https://pig.apache.org/docs/r0.9.1/basic.html#filter>`__)
-  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.html#store>`__)

La commande magique permet d’écrire un fichier ``select1.pig`` avec
l’encoding ``utf-8`` (les caractères accentuées sont possibles).

.. code:: ipython3

    %%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() ;

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 :

.. code:: ipython3

    jid = %hd_pig_submit select_azure_2015.pig
    jid




.. parsed-literal::
    {'id': 'job_1445989166328_0007'}



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é :

.. code:: ipython3

    df = %blob_ls /$PSEUDO/scripts/pig/
    df[df.name.apply(lambda x : "select" in x)]






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>last_modified</th>
          <th>content_type</th>
          <th>content_length</th>
          <th>blob_type</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>36</th>
          <td>xavierdupre/scripts/pig/select_azure.pig</td>
          <td>Thu, 29 Oct 2015 00:30:39 GMT</td>
          <td>application/octet-stream</td>
          <td>374</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>37</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log</td>
          <td>Thu, 29 Oct 2015 00:31:19 GMT</td>
          <td></td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>38</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log/exit</td>
          <td>Thu, 29 Oct 2015 00:31:19 GMT</td>
          <td>application/octet-stream</td>
          <td>3</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>39</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log/s...</td>
          <td>Thu, 29 Oct 2015 00:31:09 GMT</td>
          <td>application/octet-stream</td>
          <td>4808</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>40</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log/s...</td>
          <td>Thu, 29 Oct 2015 00:31:09 GMT</td>
          <td>application/octet-stream</td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>41</th>
          <td>xavierdupre/scripts/pig/select_azure_2015.pig</td>
          <td>Thu, 29 Oct 2015 00:45:05 GMT</td>
          <td>application/octet-stream</td>
          <td>375</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>42</th>
          <td>xavierdupre/scripts/pig/select_azure_2015.pig.log</td>
          <td>Thu, 29 Oct 2015 00:45:28 GMT</td>
          <td></td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>43</th>
          <td>xavierdupre/scripts/pig/select_azure_2015.pig....</td>
          <td>Thu, 29 Oct 2015 00:46:15 GMT</td>
          <td>application/octet-stream</td>
          <td>15812</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>44</th>
          <td>xavierdupre/scripts/pig/select_azure_2015.pig....</td>
          <td>Thu, 29 Oct 2015 00:46:15 GMT</td>
          <td>application/octet-stream</td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
      </tbody>
    </table>
    </div>



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é.

.. code:: ipython3

    st = %hd_job_status jid["id"]
    st["id"],st["percentComplete"],st["completed"],st["status"]["jobComplete"],st["status"]["state"]




.. parsed-literal::
    ('job_1445989166328_0007', '100% complete', 'done', True, 'SUCCEEDED')



On vérifie le contenu du cluster :

.. code:: ipython3

    df=%blob_ls /$PSEUDO/testensae
    df[df.last_modified.apply(lambda x : "2015" in x)]






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>last_modified</th>
          <th>content_type</th>
          <th>content_length</th>
          <th>blob_type</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>10</th>
          <td>xavierdupre/testensae/ConfLongDemo_JSI.small.w...</td>
          <td>Thu, 29 Oct 2015 00:28:30 GMT</td>
          <td></td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>11</th>
          <td>xavierdupre/testensae/ConfLongDemo_JSI.small.w...</td>
          <td>Thu, 29 Oct 2015 00:28:30 GMT</td>
          <td>application/octet-stream</td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>12</th>
          <td>xavierdupre/testensae/ConfLongDemo_JSI.small.w...</td>
          <td>Thu, 29 Oct 2015 00:28:30 GMT</td>
          <td>application/octet-stream</td>
          <td>22166</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>13</th>
          <td>xavierdupre/testensae/ConfLongDemo_JSI.small.w...</td>
          <td>Thu, 29 Oct 2015 00:46:05 GMT</td>
          <td></td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>14</th>
          <td>xavierdupre/testensae/ConfLongDemo_JSI.small.w...</td>
          <td>Thu, 29 Oct 2015 00:46:05 GMT</td>
          <td>application/octet-stream</td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>15</th>
          <td>xavierdupre/testensae/ConfLongDemo_JSI.small.w...</td>
          <td>Thu, 29 Oct 2015 00:46:04 GMT</td>
          <td>application/octet-stream</td>
          <td>22166</td>
          <td>BlockBlob</td>
        </tr>
      </tbody>
    </table>
    </div>



.. code:: ipython3

    set(df[df.last_modified.apply(lambda x : "2015" in x)]["name"])




.. parsed-literal::
    {'xavierdupre/testensae/ConfLongDemo_JSI.small.walking2015.txt',
     'xavierdupre/testensae/ConfLongDemo_JSI.small.walking2015.txt/_SUCCESS',
     'xavierdupre/testensae/ConfLongDemo_JSI.small.walking2015.txt/part-m-00000',
     'xavierdupre/testensae/ConfLongDemo_JSI.small.walking_2015.txt',
     'xavierdupre/testensae/ConfLongDemo_JSI.small.walking_2015.txt/_SUCCESS',
     'xavierdupre/testensae/ConfLongDemo_JSI.small.walking_2015.txt/part-m-00000'}



Le job s’est bien exécuté. On downloade le contenu sur l’ordinateur :

.. code:: ipython3

    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




.. parsed-literal::
    'results.txt'



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 :

.. code:: ipython3

    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




.. parsed-literal::
    'results_allfiles.txt'



.. code:: ipython3

    %lsr r.*s[.]txt






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>directory</th>
          <th>last_modified</th>
          <th>name</th>
          <th>size</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>False</td>
          <td>2015-10-29 01:46:45.425028</td>
          <td>.\results.txt</td>
          <td>21.65 Kb</td>
        </tr>
        <tr>
          <th>1</th>
          <td>False</td>
          <td>2015-10-29 01:46:46.705466</td>
          <td>.\results_allfiles.txt</td>
          <td>21.65 Kb</td>
        </tr>
      </tbody>
    </table>
    </div>



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.

.. code:: ipython3

    if os.path.exists("stderr.txt"): os.remove("stderr.txt")
    %blob_ls /$SCRIPTPIG/select_azure.pig.log






.. raw:: html

    <div>
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>name</th>
          <th>last_modified</th>
          <th>content_type</th>
          <th>content_length</th>
          <th>blob_type</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log</td>
          <td>Thu, 29 Oct 2015 00:31:19 GMT</td>
          <td></td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>1</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log/exit</td>
          <td>Thu, 29 Oct 2015 00:31:19 GMT</td>
          <td>application/octet-stream</td>
          <td>3</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>2</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log/s...</td>
          <td>Thu, 29 Oct 2015 00:31:09 GMT</td>
          <td>application/octet-stream</td>
          <td>4808</td>
          <td>BlockBlob</td>
        </tr>
        <tr>
          <th>3</th>
          <td>xavierdupre/scripts/pig/select_azure.pig.log/s...</td>
          <td>Thu, 29 Oct 2015 00:31:09 GMT</td>
          <td>application/octet-stream</td>
          <td>0</td>
          <td>BlockBlob</td>
        </tr>
      </tbody>
    </table>
    </div>



.. code:: ipython3

    df = %blob_ls /$SCRIPTPIG/select_azure_2015.pig.log
    list(df["name"])




.. parsed-literal::
    ['xavierdupre/scripts/pig/select_azure_2015.pig.log',
     'xavierdupre/scripts/pig/select_azure_2015.pig.log/exit',
     'xavierdupre/scripts/pig/select_azure_2015.pig.log/stderr',
     'xavierdupre/scripts/pig/select_azure_2015.pig.log/stdout']



L’instruction récupère l’erreur standard pour un job donné :

.. code:: ipython3

    %hd_tail_stderr jid["id"]






.. raw:: html

    <pre>

    Job DAG:
    job_1445989166328_0008


    2015-10-29 00:46:13,803 [main] INFO  org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service address: http://headnodehost:8188/ws/v1/timeline/
    2015-10-29 00:46:13,803 [main] INFO  org.apache.hadoop.yarn.client.RMProxy - Connecting to ResourceManager at headnodehost/100.89.100.164:9010
    2015-10-29 00:46:13,803 [main] INFO  org.apache.hadoop.yarn.client.AHSProxy - Connecting to Application History server at headnodehost/100.89.100.164:10200
    2015-10-29 00:46:13,881 [main] INFO  org.apache.hadoop.mapred.ClientServiceDelegate - Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server
    2015-10-29 00:46:14,068 [main] INFO  org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service address: http://headnodehost:8188/ws/v1/timeline/
    2015-10-29 00:46:14,068 [main] INFO  org.apache.hadoop.yarn.client.RMProxy - Connecting to ResourceManager at headnodehost/100.89.100.164:9010
    2015-10-29 00:46:14,068 [main] INFO  org.apache.hadoop.yarn.client.AHSProxy - Connecting to Application History server at headnodehost/100.89.100.164:10200
    2015-10-29 00:46:14,146 [main] INFO  org.apache.hadoop.mapred.ClientServiceDelegate - Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server
    2015-10-29 00:46:14,334 [main] INFO  org.apache.hadoop.yarn.client.api.impl.TimelineClientImpl - Timeline service address: http://headnodehost:8188/ws/v1/timeline/
    2015-10-29 00:46:14,334 [main] INFO  org.apache.hadoop.yarn.client.RMProxy - Connecting to ResourceManager at headnodehost/100.89.100.164:9010
    2015-10-29 00:46:14,334 [main] INFO  org.apache.hadoop.yarn.client.AHSProxy - Connecting to Application History server at headnodehost/100.89.100.164:10200
    2015-10-29 00:46:14,412 [main] INFO  org.apache.hadoop.mapred.ClientServiceDelegate - Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server
    2015-10-29 00:46:14,490 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!
    2015-10-29 00:46:14,522 [main] INFO  org.apache.pig.Main - Pig script completed in 44 seconds and 962 milliseconds (44962 ms)

    </pre><br />



Et pour tuer un job, il sufft d’appeler :

.. code:: ipython3

    %hd_job_kill <job_id>

Partie 3 : syntaxe PIG et exercices
-----------------------------------

Dans 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
~~~~~~~~~~~~~~~~~~~~~

.. code:: ipython3

    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()






.. raw:: html

    <div style="max-height:1000px;max-width:1500px;overflow:auto;">
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>activity</th>
          <th>nb</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td>      falling</td>
          <td>  2973</td>
        </tr>
        <tr>
          <th>1</th>
          <td>        lying</td>
          <td> 54480</td>
        </tr>
        <tr>
          <th>2</th>
          <td>   lying down</td>
          <td>  6168</td>
        </tr>
        <tr>
          <th>3</th>
          <td> on all fours</td>
          <td>  5210</td>
        </tr>
        <tr>
          <th>4</th>
          <td>      sitting</td>
          <td> 27244</td>
        </tr>
      </tbody>
    </table>
    </div>



Il faut maintenant le faire avec PIG.


Exercice 2 : JOIN
~~~~~~~~~~~~~~~~~

.. code:: ipython3

    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()






.. raw:: html

    <div style="max-height:1000px;max-width:1500px;overflow:auto;">
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th></th>
          <th>index</th>
          <th>sequence</th>
          <th>tag</th>
          <th>timestamp</th>
          <th>dateformat</th>
          <th>x</th>
          <th>y</th>
          <th>z</th>
          <th>activity</th>
          <th>nb</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td> 0</td>
          <td> A01</td>
          <td> 010-000-024-033</td>
          <td> 633790226051280329</td>
          <td> 27.05.2009 14:03:25:127</td>
          <td> 4.062931</td>
          <td> 1.892434</td>
          <td> 0.507425</td>
          <td> walking</td>
          <td> 32710</td>
        </tr>
        <tr>
          <th>1</th>
          <td> 1</td>
          <td> A01</td>
          <td> 020-000-033-111</td>
          <td> 633790226051820913</td>
          <td> 27.05.2009 14:03:25:183</td>
          <td> 4.291954</td>
          <td> 1.781140</td>
          <td> 1.344495</td>
          <td> walking</td>
          <td> 32710</td>
        </tr>
        <tr>
          <th>2</th>
          <td> 2</td>
          <td> A01</td>
          <td> 020-000-032-221</td>
          <td> 633790226052091205</td>
          <td> 27.05.2009 14:03:25:210</td>
          <td> 4.359101</td>
          <td> 1.826456</td>
          <td> 0.968821</td>
          <td> walking</td>
          <td> 32710</td>
        </tr>
        <tr>
          <th>3</th>
          <td> 3</td>
          <td> A01</td>
          <td> 010-000-024-033</td>
          <td> 633790226052361498</td>
          <td> 27.05.2009 14:03:25:237</td>
          <td> 4.087835</td>
          <td> 1.879999</td>
          <td> 0.466983</td>
          <td> walking</td>
          <td> 32710</td>
        </tr>
        <tr>
          <th>4</th>
          <td> 4</td>
          <td> A01</td>
          <td> 010-000-030-096</td>
          <td> 633790226052631792</td>
          <td> 27.05.2009 14:03:25:263</td>
          <td> 4.324462</td>
          <td> 2.072460</td>
          <td> 0.488065</td>
          <td> walking</td>
          <td> 32710</td>
        </tr>
      </tbody>
    </table>
    </div>



Idem, maintenant il faut le faire avec PIG.