.. _pigstreamingclouderacorrectionrst:

==========================================================================
PIG et JSON et streaming avec les données vélib - correction avec cloudera
==========================================================================


.. only:: html

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


Correction.

.. code:: ipython3

    from jyquickhelper import add_notebook_menu
    add_notebook_menu()






.. contents::
    :local:





Récupération des données
------------------------

Le code suivant télécharge les données nécessaires
`data_velib_paris_2014-11-11_22-23.zip <http://www.xavierdupre.fr/enseignement/complements/data_velib_paris_2014-11-11_22-23.zip>`__.

.. code:: ipython3

    import pyensae.datasource
    import os, datetime
    if not os.path.exists("velib"):
        os.mkdir("velib")
    files=pyensae.datasource.download_data("data_velib_paris_2014-11-11_22-23.zip", website="xdtd", whereTo="velib")
    files[:2]




.. parsed-literal::
    ['velib\\paris.2014-11-11_22-00-18.331391.txt',
     'velib\\paris.2014-11-11_22-01-17.859194.txt']



Connexion au cluster
--------------------

.. code:: ipython3

    import pyensae
    %load_ext pyensae
    %load_ext pyenbc
    from pyquickhelper.ipythonhelper import open_html_form
    params={"server":"df...fr", "username":"", "password":""}
    open_html_form(params=params,title="server + credentials", key_save="params")






.. raw:: html

    <div style="background-color:gainsboro; padding:2px; border:0px;"><b>server + credentials</b>
    <br />password <input type="password" id="paramspassword" value="" size="80" />
    <br />server <input type="text" id="paramsserver" value="df...fr" size="80" />
    <br />username <input type="text" id="paramsusername" value="" size="80" />
    <br /><button onclick="set_valueparams()">Ok</button></div>
    <script type="text/Javascript">
    function paramscallback(msg) {
       var ret = msg.content.data['text/plain'];
       $('#outparams').text(ret);
    }
    function set_valueparams(){
       command='params = {' ;
       var paramspasswordvar_value = document.getElementById('paramspassword').value;
       command += '"password":"' + paramspasswordvar_value + '",';
       var paramsservervar_value = document.getElementById('paramsserver').value;
       command += '"server":"' + paramsservervar_value + '",';
       var paramsusernamevar_value = document.getElementById('paramsusername').value;
       command += '"username":"' + paramsusernamevar_value + '",';
       command += '}';
       var kernel = IPython.notebook.kernel;
       kernel.execute(command);
    }
    </script>



.. code:: ipython3

    import pyensae
    %load_ext pyensae
    password = params["password"]
    server = params["server"]
    username = params["username"]
    client = %remote_open
    client




.. parsed-literal::
    <pyensae.remote.remote_connection.ASSHClient at 0x5b43bb0>



Exercice 1 : convertir les valeurs numériques
---------------------------------------------

Le programme suivant prend comme argument les colonnes à extraire des
fichiers textes qui sont enregistrés au format “python”.

.. code:: ipython3

    %%PYTHON stream_json.py
    
    import sys, datetime
    cols = [ _ for _ in sys.argv if ".py" not in _ ]
    for row in sys.stdin:
        row = row.strip()
        if len(row) == 0 :
            continue
        js = eval(row)
        for station in js:
            vals = [ str(station[c]) for c in cols ]
            sys.stdout.write(",".join(vals))
            sys.stdout.write("\n")
            sys.stdout.flush()

On vérifie qu’il fonctionne en lui faisant ingérer les deux lignes
suivantes :

.. code:: ipython3

    %%runpy stream_json.py name status
    [{'address': 'RUE DES CHAMPEAUX (PRES DE LA GARE ROUTIERE) - 93170 BAGNOLET', 'collect_date': datetime.datetime(2014, 11, 11, 22, 2, 18, 47270), 'lng': 2.416170724425901, 'contract_name': 'Paris', 'name': '31705 - CHAMPEAUX (BAGNOLET)', 'banking': 0, 'lat': 48.8645278209514, 'bonus': 0, 'status': 'OPEN', 'available_bikes': 1, 'last_update': datetime.datetime(2014, 11, 11, 21, 55, 22), 'number': 31705, 'available_bike_stands': 49, 'bike_stands': 50}]
    [{'address': 'RUE DES CHAMPEAUX (PRES DE LA GARE ROUTIERE) - 93170 BAGNOLET', 'collect_date': datetime.datetime(2014, 11, 11, 22, 2, 18, 47270), 'lng': 2.416170724425901, 'contract_name': 'Paris', 'name': '31705 - CHAMPEAUX (BAGNOLET)', 'banking': 0, 'lat': 48.8645278209514, 'bonus': 0, 'status': 'OPEN', 'available_bikes': 1, 'last_update': datetime.datetime(2014, 11, 11, 21, 55, 22), 'number': 31705, 'available_bike_stands': 49, 'bike_stands': 50}]






.. raw:: html

    <pre>
    31705 - CHAMPEAUX (BAGNOLET),OPEN
    31705 - CHAMPEAUX (BAGNOLET),OPEN

    </pre>



On écrit le script PIG qui utilise plus de colonnes :

.. code:: ipython3

    %%PIG json_velib.pig
    
    DEFINE pystream `python stream_json.py available_bike_stands available_bikes lat lng name status` 
       SHIP ('stream_json.py') INPUT(stdin USING PigStreaming(',')) OUTPUT (stdout USING PigStreaming(',')) ;
    
    jspy = LOAD 'velib_py/*.txt' USING PigStorage('\t') AS (arow:chararray);
    
    matrice = STREAM jspy THROUGH pystream AS 
                    (   available_bike_stands:int, 
                        available_bikes:int, 
                        lat:double, 
                        lng:double, 
                        name:chararray, 
                        status:chararray) ;
    
    STORE matrice INTO 'velib_py_results/firstjob' USING PigStorage('\t') ;

On supprime la précédente exécution si besoin puis on vérifie que le
répertoire contenant les résultats est vide :

.. code:: ipython3

    if client.dfs_exists("velib_py_results/firstjob"):
        client.dfs_rm("velib_py_results/firstjob", recursive=True)
    %dfs_mkdir velib_py_results




.. parsed-literal::
    ("Moved: 'hdfs://nameservice1/user/xavierdupre/velib_py_results/firstjob' to trash at: hdfs://nameservice1/user/xavierdupre/.Trash/Current\n",
     '14/11/23 20:35:18 INFO fs.TrashPolicyDefault: Namenode trash configuration: Deletion interval = 1440 minutes, Emptier interval = 0 minutes.\n')



.. code:: ipython3

    %dfs_ls velib_py_results/






.. 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>attributes</th>
          <th>code</th>
          <th>alias</th>
          <th>folder</th>
          <th>size</th>
          <th>date</th>
          <th>time</th>
          <th>name</th>
        </tr>
      </thead>
      <tbody>
      </tbody>
    </table>
    </div>



.. code:: ipython3

    %pig_submit json_velib.pig stream_json.py -r redirection






.. raw:: html

    <pre>

    </pre>



.. code:: ipython3

    %remote_cmd tail redirection.pig.err






.. raw:: html

    <pre>
    Total bytes written : 4611956
    Spillable Memory Manager spill count : 0
    Total bags proactively spilled: 0
    Total records proactively spilled: 0

    Job DAG:
    job_1414491244634_0119


    2014-11-23 20:36:50,180 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!

    </pre>



On récupère les informations qu’on affiche sous forme de dataframe :

.. code:: ipython3

    if os.path.exists("velib_exo1.txt") : os.remove("velib_exo1.txt")
    client.download_cluster("velib_py_results/firstjob","velib_exo1.txt", merge=True)




.. parsed-literal::
    'velib_py_results/firstjob'



.. code:: ipython3

    import pandas
    df = pandas.read_csv("velib_hd.txt", sep="\t",names=["available_bike_stands","available_bikes","lat","lng","name","status"])
    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>available_bike_stands</th>
          <th>available_bikes</th>
          <th>lat</th>
          <th>lng</th>
          <th>name</th>
          <th>status</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td> 47</td>
          <td>  3</td>
          <td> 48.864528</td>
          <td> 2.416171</td>
          <td>   31705 - CHAMPEAUX (BAGNOLET)</td>
          <td> OPEN</td>
        </tr>
        <tr>
          <th>1</th>
          <td>  5</td>
          <td> 28</td>
          <td> 48.872420</td>
          <td> 2.348395</td>
          <td> 10042 - POISSONNIÈRE - ENGHIEN</td>
          <td> OPEN</td>
        </tr>
        <tr>
          <th>2</th>
          <td> 42</td>
          <td>  1</td>
          <td> 48.882149</td>
          <td> 2.319860</td>
          <td>             08020 - METRO ROME</td>
          <td> OPEN</td>
        </tr>
        <tr>
          <th>3</th>
          <td>  5</td>
          <td> 31</td>
          <td> 48.868217</td>
          <td> 2.330494</td>
          <td>         01022 - RUE DE LA PAIX</td>
          <td> OPEN</td>
        </tr>
        <tr>
          <th>4</th>
          <td> 20</td>
          <td>  5</td>
          <td> 48.893269</td>
          <td> 2.412716</td>
          <td>     35014 - DE GAULLE (PANTIN)</td>
          <td> OPEN</td>
        </tr>
      </tbody>
    </table>
    </div>



Exercice 2 : stations fermées
-----------------------------

Les stations fermées ne le sont pas tout le temps. On veut calculer le
ratio minutes/stations fermées / total des minutes/stations. Le script
python permettant de lire les données ne change pas, il suffit juste de
déclarer les sorties numériques comme telles. Quelques détails sur les
tables :


-  ``jspy`` : contient les données brutes dans une liste de fichiers
-  ``matrice`` : d’après le job qui précède, la table contient une ligne
   par stations et par minute, chaque ligne décrit le status de la
   station
-  ``grstation`` : table ``matrice`` groupée par status
-  ``fermees`` : pour chaque groupe, on aggrégé le nombre de minutes
   multipliés par le nombre de vélos
-  ``gr*,dist*`` : distribution du nombre de stations (Y) en fonction du
   nombre de vélos ou places disponibles

En cas d’exécution précédentes :

.. code:: ipython3

    %dfs_rmr velib_py_results/fermees.txt
    %dfs_rmr velib_py_results/distribution_bikes.txt
    %dfs_rmr velib_py_results/distribution_stands.txt




.. parsed-literal::
    ("Moved: 'hdfs://nameservice1/user/xavierdupre/velib_py_results/distribution_stands.txt' to trash at: hdfs://nameservice1/user/xavierdupre/.Trash/Current\n",
     '14/11/23 21:33:49 INFO fs.TrashPolicyDefault: Namenode trash configuration: Deletion interval = 1440 minutes, Emptier interval = 0 minutes.\n')



.. code:: ipython3

    %%PIG json_fermee.pig
    
    DEFINE pystream `python stream_json.py available_bike_stands available_bikes lat lng name status` 
       SHIP ('stream_json.py') INPUT(stdin USING PigStreaming(',')) OUTPUT (stdout USING PigStreaming(',')) ;
    
    jspy = LOAD 'velib_py/*.txt' USING PigStorage('\t') AS (arow:chararray);
    
    matrice = STREAM jspy THROUGH pystream AS 
                    (   available_bike_stands:int, 
                        available_bikes:int, 
                        lat:double, 
                        lng:double, 
                        name:chararray, 
                        status:chararray) ;
                    
    grstation = GROUP matrice BY status ;  
    
    fermees = FOREACH grstation GENERATE
                     group
                    ,SUM(matrice.available_bikes) AS available_bikes
                    ,SUM(matrice.available_bike_stands) AS available_bike_stands 
                    ;
    
    gr_av   = GROUP matrice BY available_bikes ;
    dist_av = FOREACH gr_av GENERATE group, COUNT(matrice) ;
    gr_pl   = GROUP matrice BY available_bike_stands ;
    dist_pl = FOREACH gr_pl GENERATE group, COUNT(matrice) ;
    
    STORE fermees INTO 'velib_py_results/fermees.txt' USING PigStorage('\t') ;
    STORE dist_av INTO 'velib_py_results/distribution_bikes.txt' USING PigStorage('\t') ;
    STORE dist_pl INTO 'velib_py_results/distribution_stands.txt' USING PigStorage('\t') ;

.. code:: ipython3

    %pig_submit json_fermee.pig stream_json.py -r redirection






.. raw:: html

    <pre>

    </pre>



.. code:: ipython3

    %remote_cmd tail redirection.pig.err 






.. raw:: html

    <pre>
    Total bytes written : 1009
    Spillable Memory Manager spill count : 0
    Total bags proactively spilled: 0
    Total records proactively spilled: 0

    Job DAG:
    job_1414491244634_0124


    2014-11-23 21:35:28,238 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!

    </pre>



.. code:: ipython3

    %dfs_ls velib_py_results






.. 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>attributes</th>
          <th>code</th>
          <th>alias</th>
          <th>folder</th>
          <th>size</th>
          <th>date</th>
          <th>time</th>
          <th>name</th>
          <th>isdir</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 0</td>
          <td> 2014-11-23</td>
          <td> 20:54</td>
          <td>  velib_py_results/distribution_bikes.txt</td>
          <td> True</td>
        </tr>
        <tr>
          <th>1</th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 0</td>
          <td> 2014-11-23</td>
          <td> 20:54</td>
          <td> velib_py_results/distribution_stands.txt</td>
          <td> True</td>
        </tr>
        <tr>
          <th>2</th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 0</td>
          <td> 2014-11-23</td>
          <td> 20:36</td>
          <td>                velib_py_results/firstjob</td>
          <td> True</td>
        </tr>
      </tbody>
    </table>
    </div>



.. code:: ipython3

    if os.path.exists("distribution_bikes.txt") : os.remove("distribution_bikes.txt")
    %remote_down_cluster_merge velib_py_results/distribution_bikes.txt distribution_bikes.txt




.. parsed-literal::
    'distribution_bikes.txt'



.. code:: ipython3

    import pandas
    df = pandas.read_csv("distribution_bikes.txt", sep="\t", names=["nb_velos", "nb_stations_minutes"])
    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>nb_velos</th>
          <th>nb_stations_minutes</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td> 0</td>
          <td> 8445</td>
        </tr>
        <tr>
          <th>1</th>
          <td> 1</td>
          <td> 6589</td>
        </tr>
        <tr>
          <th>2</th>
          <td> 2</td>
          <td> 4825</td>
        </tr>
        <tr>
          <th>3</th>
          <td> 3</td>
          <td> 3793</td>
        </tr>
        <tr>
          <th>4</th>
          <td> 4</td>
          <td> 2839</td>
        </tr>
      </tbody>
    </table>
    </div>



.. code:: ipython3

    df.plot(x="nb_velos",y="nb_stations_minutes",kind="bar",figsize=(16,4))




.. parsed-literal::
    <matplotlib.axes.AxesSubplot at 0x9f3c4d0>




.. image:: pig_streaming_cloudera_correction_30_1.png


.. code:: ipython3

    if os.path.exists("distribution_stands.txt") : os.remove("distribution_stands.txt")
    %remote_down_cluster_merge velib_py_results/distribution_stands.txt distribution_stands.txt




.. parsed-literal::
    'distribution_stands.txt'



.. code:: ipython3

    df = pandas.read_csv("distribution_stands.txt", sep="\t", names=["nb_places", "nb_stations_minutes"])
    df.plot(x="nb_places",y="nb_stations_minutes",kind="bar",figsize=(16,4))




.. parsed-literal::
    <matplotlib.axes.AxesSubplot at 0xa04d390>




.. image:: pig_streaming_cloudera_correction_32_1.png


.. code:: ipython3

    if os.path.exists("fermees.txt") : os.remove("fermees.txt")
    %remote_down_cluster_merge velib_py_results/fermees.txt fermees.txt




.. parsed-literal::
    'fermees.txt'



.. code:: ipython3

    df = pandas.read_csv("fermees.txt", sep="\t", names=["status", "nb_velos_stations_minutes", "nb_places_stations_minutes"])
    df=df.set_index("status")
    df = df.T
    df["%close"] = df.CLOSED / (df.CLOSED + df.OPEN)
    df






.. raw:: html

    <div style="max-height:1000px;max-width:1500px;overflow:auto;">
    <table border="1" class="dataframe">
      <thead>
        <tr style="text-align: right;">
          <th>status</th>
          <th>OPEN</th>
          <th>CLOSED</th>
          <th>%close</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>nb_velos_stations_minutes</th>
          <td> 1048654</td>
          <td> 3060</td>
          <td> 0.002910</td>
        </tr>
        <tr>
          <th>nb_places_stations_minutes</th>
          <td> 1255146</td>
          <td>  120</td>
          <td> 0.000096</td>
        </tr>
      </tbody>
    </table>
    </div>



Ce dernier tableau n’est vrai que dans la mesure où les nombres
reportées sont fiables lorsque les stations sont fermées.

Exercice 3 : stations fermées, journée complète
-----------------------------------------------

Appliquer cela à une journée complète revient à lancer le même job sur
des données plus grandes. On verra bientôt commencer éviter de recopier
le job une seconde fois (introduisant une source potentielle d’erreur).

.. code:: ipython3

    %dfs_ls /user/xavierdupre/






.. 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>attributes</th>
          <th>code</th>
          <th>alias</th>
          <th>folder</th>
          <th>size</th>
          <th>date</th>
          <th>time</th>
          <th>name</th>
          <th>isdir</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0 </th>
          <td> drwx------</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-23</td>
          <td> 17:55</td>
          <td>                          /user/xavierdupre/.Trash</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>1 </th>
          <td> drwx------</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-23</td>
          <td> 21:41</td>
          <td>                        /user/xavierdupre/.staging</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>2 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 132727</td>
          <td> 2014-11-16</td>
          <td> 02:37</td>
          <td> /user/xavierdupre/ConfLongDemo_JSI.small.examp...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>3 </th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-16</td>
          <td> 02:38</td>
          <td> /user/xavierdupre/ConfLongDemo_JSI.small.examp...</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>4 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 461444</td>
          <td> 2014-11-20</td>
          <td> 01:33</td>
          <td> /user/xavierdupre/paris.2014-11-11_22-00-18.33...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>5 </th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-20</td>
          <td> 23:43</td>
          <td>                        /user/xavierdupre/unitest2</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>6 </th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-23</td>
          <td> 21:42</td>
          <td>                        /user/xavierdupre/unittest</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>7 </th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-23</td>
          <td> 21:41</td>
          <td>                       /user/xavierdupre/unittest2</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>8 </th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-20</td>
          <td> 01:53</td>
          <td>                      /user/xavierdupre/velib_1hjs</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>9 </th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-21</td>
          <td> 01:17</td>
          <td>                        /user/xavierdupre/velib_py</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>10</th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-23</td>
          <td> 21:34</td>
          <td>                /user/xavierdupre/velib_py_results</td>
          <td>  True</td>
        </tr>
        <tr>
          <th>11</th>
          <td> drwxr-xr-x</td>
          <td> -</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> 2014-11-21</td>
          <td> 11:08</td>
          <td>              /user/xavierdupre/velib_several_days</td>
          <td>  True</td>
        </tr>
      </tbody>
    </table>
    </div>



.. code:: ipython3

    %%PIG json_fermee.pig
    
    DEFINE pystream `python stream_json.py available_bike_stands available_bikes lat lng name status` 
       SHIP ('stream_json.py') INPUT(stdin USING PigStreaming(',')) OUTPUT (stdout USING PigStreaming(',')) ;
    
    jspy = LOAD '/user/xavierdupre/velib_several_days/*.txt' USING PigStorage('\t') AS (arow:chararray);
    
    matrice = STREAM jspy THROUGH pystream AS 
                    (   available_bike_stands:int, 
                        available_bikes:int, 
                        lat:double, 
                        lng:double, 
                        name:chararray, 
                        status:chararray) ;
                    
    grstation = GROUP matrice BY status ;  
    
    fermees = FOREACH grstation GENERATE
                     group
                    ,SUM(matrice.available_bikes) AS available_bikes
                    ,SUM(matrice.available_bike_stands) AS available_bike_stands 
                    ;
    
    gr_av   = GROUP matrice BY available_bikes ;
    dist_av = FOREACH gr_av GENERATE group, COUNT(matrice) ;
    gr_pl   = GROUP matrice BY available_bike_stands ;
    dist_pl = FOREACH gr_pl GENERATE group, COUNT(matrice) ;
    
    STORE fermees INTO 'velib_py_results_3days/fermees.txt' USING PigStorage('\t') ;
    STORE dist_av INTO 'velib_py_results_3days/distribution_bikes.txt' USING PigStorage('\t') ;
    STORE dist_pl INTO 'velib_py_results_3days/distribution_stands.txt' USING PigStorage('\t') ;

.. code:: ipython3

    %pigsubmit json_fermee.pig stream_json.py redirection






.. raw:: html

    <pre>

    </pre>



.. code:: ipython3

    %remote_cmd tail redirection.err






.. raw:: html

    <pre>
    Total bytes written : 1306
    Spillable Memory Manager spill count : 0
    Total bags proactively spilled: 0
    Total records proactively spilled: 0

    Job DAG:
    job_1414491244634_0128


    2014-11-23 21:55:06,336 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!

    </pre>



.. code:: ipython3

    %dfs_ls velib_py_results_3days/*






.. 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>attributes</th>
          <th>code</th>
          <th>alias</th>
          <th>folder</th>
          <th>size</th>
          <th>date</th>
          <th>time</th>
          <th>name</th>
          <th>isdir</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>   0</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_bikes.txt/...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 210</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_bikes.txt/...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>2 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 208</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_bikes.txt/...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>3 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 216</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_bikes.txt/...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>4 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>   0</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_stands.txt...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>5 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 210</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_stands.txt...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>6 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 208</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_stands.txt...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>7 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 210</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td> velib_py_results_3days/distribution_stands.txt...</td>
          <td> False</td>
        </tr>
        <tr>
          <th>8 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>   0</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td>       velib_py_results_3days/fermees.txt/_SUCCESS</td>
          <td> False</td>
        </tr>
        <tr>
          <th>9 </th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>  20</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td>   velib_py_results_3days/fermees.txt/part-r-00000</td>
          <td> False</td>
        </tr>
        <tr>
          <th>10</th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>   0</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td>   velib_py_results_3days/fermees.txt/part-r-00001</td>
          <td> False</td>
        </tr>
        <tr>
          <th>11</th>
          <td> -rw-r--r--</td>
          <td> 3</td>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>  24</td>
          <td> 2014-11-23</td>
          <td> 21:54</td>
          <td>   velib_py_results_3days/fermees.txt/part-r-00002</td>
          <td> False</td>
        </tr>
      </tbody>
    </table>
    </div>



On voit que le job a été distribué sur trois machines.

Exercice 4 : astuces
--------------------

Les erreurs de PIG ne sont pas très explicite surtout si elles se
produisent dans le script python. Un moyen simple de débugger est
d’attraper les exceptions produites par python et de les récupérer sous
PIG (le reste du job est enlevé). On peut tout-à-fait imaginer ajouter
la version de python installée sur le cluster ainsi que la liste de
modules

.. code:: ipython3

    %%PYTHON stream_json_err.py
    
    import sys, os
    for row in sys.stdin:
        sys.stdout.write(sys.executable + "\n")
        sys.stdout.write(str(sys.version) + "\n")
        sys.stdout.write(row + "\n")
        # pour obtenir les variables d'environnement
        #for k in os.environ:
        #    sys.stdout.write("%s=%s\n" % (k,os.environ[k]))
        sys.stdout.flush()

.. code:: ipython3

    %%runpy stream_json_err.py
    n'importe quoi






.. raw:: html

    <pre>
    C:\Python34\python.exe
    3.4.1 (v3.4.1:c0e311e010fc, May 18 2014, 10:38:22) [MSC v.1600 32 bit (Intel)]
    n'importe quoi

    </pre>



.. code:: ipython3

    %%PIG test.pig
    
    DEFINE pystream `python stream_json_err.py` 
       SHIP ('stream_json_err.py') INPUT(stdin USING PigStreaming(',')) OUTPUT (stdout USING PigStreaming(',')) ;
    
    jspy = LOAD 'velib_py/*.txt' USING PigStorage('\t') AS (arow:chararray);
    one = LIMIT jspy 1 ;
    info = STREAM one THROUGH pystream AS (row:chararray) ;
    STORE info INTO 'python_info2.txt' ;

.. code:: ipython3

    %pig_submit test.pig stream_json_err.py -r redirection






.. raw:: html

    <pre>

    </pre>



.. code:: ipython3

    %remote_cmd tail redirection.pig.err -n 5






.. raw:: html

    <pre>
    2014-11-23 22:07:20,132 [main] INFO  org.apache.hadoop.ipc.Client - Retrying connect to server: dn05.dzr323.dza.datazoomr.com/10.58.223.25:59338. Already tried 1 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=3, sleepTime=1000 MILLISECONDS)
    2014-11-23 22:07:21,134 [main] INFO  org.apache.hadoop.ipc.Client - Retrying connect to server: dn05.dzr323.dza.datazoomr.com/10.58.223.25:59338. Already tried 2 time(s); retry policy is RetryUpToMaximumCountWithFixedSleep(maxRetries=3, sleepTime=1000 MILLISECONDS)
    2014-11-23 22:07:21,251 [main] INFO  org.apache.hadoop.mapred.ClientServiceDelegate - Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server
    2014-11-23 22:07:21,471 [main] INFO  org.apache.hadoop.mapred.ClientServiceDelegate - Application state is completed. FinalApplicationStatus=SUCCEEDED. Redirecting to job history server
    2014-11-23 22:07:21,574 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!

    </pre>



.. code:: ipython3

    if os.path.exists("python_info2.txt") : os.remove("python_info2.txt")
    %remote_down_cluster_merge python_info2.txt python_info2.txt




.. parsed-literal::
    'python_info2.txt'



.. code:: ipython3

    %head python_info2.txt






.. raw:: html

    <pre>
    /usr/bin/python
    2.6.6 (r266:84292
    [GCC 4.4.7 20120313 (Red Hat 4.4.7-3)]
    [{'address': 'RUE DES CHAMPEAUX (PRES DE LA GARE ROUTIERE) - 93170 BAGNOLET'


    </pre>