.. _pigstreamingrst:

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


.. only:: html

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


Manipulation de données JSON en Map/Reduce avec
`PIG <https://pig.apache.org/>`__.

.. code:: ipython3

    from jyquickhelper import add_notebook_menu
    add_notebook_menu()






.. contents::
    :local:





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

Les données ne se présentent pas toujours sous forme de tables. Elles
sont parfois non structurées comme c’est le cas des données vélib qui
sont décrites au format
`JSON <http://fr.wikipedia.org/wiki/JavaScript_Object_Notation>`__. Le
premier jeu de données est le décompte des vélos et places disponibles
toutes les minutes le 11 novembre 2014 entre 22h et 23h à Paris. Ces
données ont été accumulées grâce au programme `collect Velib
data <http://www.xavierdupre.fr/app/pyensae/helpsphinx/all_example.html#collect-velib-data>`__.

.. code:: ipython3

    import pyensae.datasource
    %load_ext pyensae
    %load_ext pyenbc
    import os, datetime

.. code:: ipython3

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



On regarde un extrait :

.. code:: ipython3

    with open("velib/paris.2014-11-11_22-00-18.331391.txt","r",encoding="utf-8")as f :
        text = f.read()
    text[:300] + "..."




.. parsed-literal::
    "[{'address': 'RUE DES CHAMPEAUX (PRES DE LA GARE ROUTIERE) - 93170 BAGNOLET', 'collect_date': datetime.datetime(2014, 11, 11, 22, 0, 18, 628226), 'lng': 2.416170724425901, 'contract_name': 'Paris', 'name': '31705 - CHAMPEAUX (BAGNOLET)', 'banking': 0, 'lat': 48.8645278209514, 'bonus': 0, 'status': '..."



Les données ont été sauvegardées au format Python si bien que la
fonction
`eval <https://docs.python.org/3.4/library/functions.html#eval>`__
suffit à les récupérer.

.. code:: ipython3

    data = eval(text)
    data[:2]




.. parsed-literal::
    [{'name': '31705 - CHAMPEAUX (BAGNOLET)',
      'available_bikes': 1,
      'status': 'OPEN',
      'number': 31705,
      'lng': 2.416170724425901,
      'available_bike_stands': 49,
      'contract_name': 'Paris',
      'address': 'RUE DES CHAMPEAUX (PRES DE LA GARE ROUTIERE) - 93170 BAGNOLET',
      'last_update': datetime.datetime(2014, 11, 11, 21, 55, 22),
      'lat': 48.8645278209514,
      'bike_stands': 50,
      'collect_date': datetime.datetime(2014, 11, 11, 22, 0, 18, 628226),
      'bonus': 0,
      'banking': 0},
     {'name': '10042 - POISSONNIÈRE - ENGHIEN',
      'available_bikes': 32,
      'status': 'OPEN',
      'number': 10042,
      'lng': 2.348395236282807,
      'available_bike_stands': 1,
      'contract_name': 'Paris',
      'address': "52 RUE D'ENGHIEN / ANGLE RUE DU FAUBOURG POISSONIERE - 75010 PARIS",
      'last_update': datetime.datetime(2014, 11, 11, 21, 59, 5),
      'lat': 48.87242006305313,
      'bike_stands': 33,
      'collect_date': datetime.datetime(2014, 11, 11, 22, 0, 18, 628226),
      'bonus': 0,
      'banking': 0}]



Et on peut les rassembler dans un DataFrame :

.. code:: ipython3

    import pandas
    df = pandas.DataFrame(data)
    df.head(n=2)






.. 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>address</th>
          <th>available_bike_stands</th>
          <th>available_bikes</th>
          <th>banking</th>
          <th>bike_stands</th>
          <th>bonus</th>
          <th>collect_date</th>
          <th>contract_name</th>
          <th>last_update</th>
          <th>lat</th>
          <th>lng</th>
          <th>name</th>
          <th>number</th>
          <th>status</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th>0</th>
          <td> RUE DES CHAMPEAUX (PRES DE LA GARE ROUTIERE) -...</td>
          <td> 49</td>
          <td>  1</td>
          <td> 0</td>
          <td> 50</td>
          <td> 0</td>
          <td>2014-11-11 22:00:18.628226</td>
          <td> Paris</td>
          <td>2014-11-11 21:55:22</td>
          <td> 48.864528</td>
          <td> 2.416171</td>
          <td>   31705 - CHAMPEAUX (BAGNOLET)</td>
          <td> 31705</td>
          <td> OPEN</td>
        </tr>
        <tr>
          <th>1</th>
          <td> 52 RUE D'ENGHIEN / ANGLE RUE DU FAUBOURG POISS...</td>
          <td>  1</td>
          <td> 32</td>
          <td> 0</td>
          <td> 33</td>
          <td> 0</td>
          <td>2014-11-11 22:00:18.628226</td>
          <td> Paris</td>
          <td>2014-11-11 21:59:05</td>
          <td> 48.872420</td>
          <td> 2.348395</td>
          <td> 10042 - POISSONNIÈRE - ENGHIEN</td>
          <td> 10042</td>
          <td> OPEN</td>
        </tr>
      </tbody>
    </table>
    </div>



Chaque ligne donne le nombre de places ``available_bike_stands`` et le
nombre de vélos ``available_bikes`` disponibles à la date
``collect_date``. La colone ``last_update`` contient l’heure du dernier
retrait ou de la dernière pose de vélo à cette station. Il y plus d’un
millier de stations à Paris (voir
`carte <http://www.velib.paris.fr/Plan-stations>`__).

.. code:: ipython3

    df.shape




.. parsed-literal::
    (1229, 14)



.. code:: ipython3

    df.plot(x="lng",y="lat",style=".",xlim=[2.20, 2.50])




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




.. image:: pig_streaming_13_1.png


L’objectif est de calculer des statistiques sur ce petit jeu de données
puis de calculer ces mêmes statistiques sur des jeux plus grands. Et on
va essayer de ne pas transformer les données localement. Elles vont être
uploadés sur le cluster au format JSON.

Conversion des données en streaming
-----------------------------------

Le format
`JSON <http://fr.wikipedia.org/wiki/JavaScript_Object_Notation>`__ telle
que l’accepte PIG stipule que chaque ligne (ou observation) doit suivre
le format ``{ id1: JSON, id2:valeur2, ... }``. Les chaînes de caractères
doivent être entourées de guillements ``"``. Il faudrait donc modifier
les fichiers précédents de telle sorte qu’ils suivent ce schéma. On
pourrait faire quelque chose comme ceci :

.. code:: ipython3

    import json
    
    class DateTimeEncoder(json.JSONEncoder):
        def default(self, obj):
            if isinstance(obj, datetime.datetime):
                encoded_object = "%04d-%02d-%02dT%02d:%02d:%02d"% (obj.timetuple()[:6] )
            else:
                encoded_object =json.JSONEncoder.default(self, obj)
            return encoded_object
    
    files = [ os.path.join("velib",_) for _ in os.listdir("velib") if "paris" in _ and _.endswith(".txt") ]
    for f in files :
        print("*****",f)
        with open(f, "r", encoding="utf8") as h:
            for row in h:
                js = eval(row)
                sjs = json.dumps( { "minute":js }, cls = DateTimeEncoder )  # essayer sans le paramètre cls pour
                                                                            # voir l'erreur que cela produit
                print(sjs [:400] + "...")
        break


.. parsed-literal::
    ***** velib\paris.2014-11-11_22-00-18.331391.txt
    {"minute": [{"name": "31705 - CHAMPEAUX (BAGNOLET)", "available_bikes": 1, "status": "OPEN", "number": 31705, "lng": 2.416170724425901, "available_bike_stands": 49, "contract_name": "Paris", "address": "RUE DES CHAMPEAUX (PRES DE LA GARE ROUTIERE) - 93170 BAGNOLET", "last_update": "2014-11-11T21:55:22", "lat": 48.8645278209514, "bike_stands": 50, "collect_date": "2014-11-11T22:00:18", "bonus": 0, ...


Toutefois, cette conversion pourrait tout aussi bien se produire sur le
cluster. C’est ce qu’on va implémenter avec l’instruction
`STREAM <http://pig.apache.org/docs/r0.8.1/piglatin_ref2.html#STREAM>`__.
Celle-ci permet d’insérer du code écrit en n’importe quel langage dans
un script PIG.

Côté python, le **streaming** se résume à l’écriture d’un programme qui
écoute l’entrée standard et retourne des résultats sur la sorties
standard. Il faut écrire selon une syntaxe acceptée par Python 2 et 3.
Ce notebook marche avec Python 3, la version sur le cluster est le plus
souvent 2.7. Cela signifie aussi que les modules en local ne sont pas
forcément installés sur le cluster.

.. 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 teste sur quelques exemples que le script fonctionne :

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



Ca fonctionne. On passe au cluster maintenant.

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

Dans un premier temps, on prend le cluster
`Cloudera <http://www.xavierdupre.fr/app/ensae_teaching_cs/helpsphinx/notebooks/td3a_cenonce_session6.html#p2>`__.
Il faut exécuter ce script pour pouvoir notifier au notebook que la
variable ``params`` existe.

.. code:: ipython3

    import pyensae
    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 0x7bdbbb0>



Upload version
--------------

On créé un répertoire sur le cluster :

.. code:: ipython3

    %dfs_mkdir velib_py




.. parsed-literal::
    ('', '')



.. code:: ipython3

    %dfs_ls .






.. 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-21</td>
          <td> 01:05</td>
          <td>                                      .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-21</td>
          <td> 01:06</td>
          <td>                                    .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>          ConfLongDemo_JSI.small.example.txt</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> ConfLongDemo_JSI.small.example2.walking.txt</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>        paris.2014-11-11_22-00-18.331391.txt</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>                                    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-20</td>
          <td> 22:29</td>
          <td>                                    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-21</td>
          <td> 01:05</td>
          <td>                                   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>                                  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:06</td>
          <td>                                    velib_py</td>
          <td>  True</td>
        </tr>
      </tbody>
    </table>
    </div>



.. code:: ipython3

    files = [ os.path.join("velib",_) for _ in os.listdir("velib") if "paris" in _ and _.endswith(".txt")]
    for i,f in enumerate(files[51:]) :
        if i % 10 == 0 : print(i,"/",len(files),f)
        filename = os.path.split(f)[-1]
        f = os.path.abspath(f)
        client.upload_cluster(f, "velib_py")


.. parsed-literal::
    0 / 61 velib\paris.2014-11-11_22-51-17.300775.txt


On crée le répertoire de résultats :

.. code:: ipython3

    %dfs_mkdir velib_py_results




.. parsed-literal::
    ('', '')



PIG, pseudo JSON et streaming
-----------------------------

On a créé plus haut le script python capable d’interpréter du python
puis d’extraire les informations voulues. On va maintenant dire à PIG
d’envoyer chaque ligne qu’il va charger (instruction ``LOAD``) au script
python (instruction ``STREAM``). On lui dit de récuper les informations
selon un certain format.

La seconde ligne ``DEFINE`` définit la commande utilisée. Elle précise
également que le script ``pystream.py`` devra être envoyé à toutes les
machines du cluster participant au job. Et il faut attention, cette
ligne ne comporte pas que des apostrophes, le caractère **’ incliné**
est très important.

Les premiers essais de streaming sont souvent douloureux.

.. 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:chararray, 
                        available_bikes:chararray, 
                        lat:double, 
                        lng:double, 
                        name:chararray, 
                        status:chararray) ;
    
    STORE matrice INTO 'velib_py_results/firstjob' USING PigStorage('\t') ;

Pour supprimer les précédents résultats :

.. 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/21 01:50:26 INFO fs.TrashPolicyDefault: Namenode trash configuration: Deletion interval = 1440 minutes, Emptier interval = 0 minutes.\n')



On exécute le job :

.. code:: ipython3

    %pig_submit json_velib.pig stream_json.py -r redirection






.. raw:: html

    <pre>

    </pre>



.. code:: ipython3

    %remote_ls .






.. 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></th>
          <th>attributes</th>
          <th>code</th>
          <th>alias</th>
          <th>folder</th>
          <th>size</th>
          <th>unit</th>
          <th>name</th>
          <th>isdir</th>
        </tr>
      </thead>
      <tbody>
        <tr>
          <th rowspan="10" valign="top">-rw-rw-r--</th>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:43</td>
          <td>                                dummy</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>    650</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:50</td>
          <td>                       json_velib.pig</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td> 523646</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:15</td>
          <td> paris.2014-11-11_22-50-17.777867.txt</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>   3077</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:24</td>
          <td>                pig_1416529443864.log</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>   3297</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:37</td>
          <td>                pig_1416530241713.log</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>    672</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:43</td>
          <td>                         pystream.pig</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>    382</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:43</td>
          <td>                          pystream.py</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>    860</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:50</td>
          <td>                      redirection.err</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>      0</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:50</td>
          <td>                      redirection.out</td>
          <td> False</td>
        </tr>
        <tr>
          <th>1</th>
          <td> xavierdupre</td>
          <td> xavierdupre</td>
          <td>    356</td>
          <td> Nov</td>
          <td> 21</td>
          <td> 01:50</td>
          <td>                       stream_json.py</td>
          <td> False</td>
        </tr>
      </tbody>
    </table>
    </div>



S’il se produit des erreurs, il est recommandé d’afficher plus de lignes
:

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


    2014-11-21 01:51:16,166 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!

    </pre>



On vérifie que tout s’est bien passé. La taille devrait être équivalent
à l’entrée.

.. 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-21</td>
          <td> 01:51</td>
          <td> velib_py_results/firstjob</td>
          <td> True</td>
        </tr>
      </tbody>
    </table>
    </div>



.. code:: ipython3

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




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



.. code:: ipython3

    %head velib_hd.txt -n 5






.. raw:: html

    <pre>
    47	3	48.864527821	2.41617072443	31705 - CHAMPEAUX (BAGNOLET)	OPEN
    5	28	48.8724200631	2.34839523628	10042 - POISSONNIÈRE - ENGHIEN	OPEN
    42	1	48.8821489456	2.31986005477	08020 - METRO ROME	OPEN
    5	31	48.8682170168	2.3304935114	01022 - RUE DE LA PAIX	OPEN
    20	5	48.8932686647	2.41271573339	35014 - DE GAULLE (PANTIN)	OPEN

    </pre>



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



.. code:: ipython3

    df.shape




.. parsed-literal::
    (73740, 6)



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


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

Les stations fermées ne le sont pas tout le temps. On veut calculer le
ratio vélo/minute/station fermées / total des vélo/minute/station.


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

Appliquer cela à une journée complète.


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