.. _pigclouderacorrectionrst:

=============================================
Map/Reduce avec PIG sur cloudera - correction
=============================================


.. only:: html

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


Correction.

.. code:: ipython3

    from jyquickhelper import add_notebook_menu
    add_notebook_menu()






.. contents::
    :local:





Données
-------

On considère 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>`__
qui ont déjà été récupéré avec le notebook de l’énoncé.

.. code:: ipython3

    import os
    if "CRTERALAB" in os.environ:
        spl = os.environ["CRTERALAB"].split("**")
        params=dict(server=spl[0], password=spl[1], username=spl[2])
        r = dict
    else:
        from pyquickhelper.ipythonhelper import open_html_form
        params={"server":"ws...fr", "username":"x...e", "password":""}
        r = open_html_form(params=params,title="server + credentials", key_save="params")
    r






.. 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="...fr" size="80" />
    <br />username <input type="text" id="paramsusername" value="x..e" 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

    password = params["password"]
    server = params["server"]
    username = params["username"]

.. code:: ipython3

    %load_ext pyensae
    %load_ext pyenbc
    %remote_open




.. parsed-literal::
    <pyensae.remote.ssh_remote_connection.ASSHClient at 0xa36fc88>



.. raw:: html

   <h3 id="exo1">

Exercice 1 : GROUP BY

.. raw:: html

   </h3>

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

.. code:: ipython3

    %%PIG solution_groupby.pig
    
    myinput = LOAD 'ConfLongDemo_JSI.small.example.txt' 
              using PigStorage(',') 
              AS (index:long, sequence, tag, timestamp:long, dateformat, x:double,y:double, z:double, activity) ;
    
    gr = GROUP myinput BY activity ;
    avgact = FOREACH gr GENERATE group, COUNT(myinput) ; 
    
    STORE avgact INTO 'ConfLongDemo_JSI.small.group.txt' USING PigStorage() ;

.. code:: ipython3

    %pig_submit solution_groupby.pig -r groupby.redirection






.. raw:: html

    <pre>

    </pre>



.. code:: ipython3

    %remote_cmd tail groupby.redirection.err






.. raw:: html

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

    Job DAG:
    job_1444669880271_0038


    2015-10-29 01:10:53,383 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!

    </pre>



.. code:: ipython3

    %remote_cmd hdfs dfs -ls ConfLongDemo_JSI.small.group.txt






.. raw:: html

    <pre>
    Found 2 items
    -rw-r--r--   3 xavierdupre xavierdupre          0 2015-10-29 01:10 ConfLongDemo_JSI.small.group.txt/_SUCCESS
    -rw-r--r--   3 xavierdupre xavierdupre         89 2015-10-29 01:10 ConfLongDemo_JSI.small.group.txt/part-r-00000

    </pre>



.. code:: ipython3

    %remote_cmd hdfs dfs -tail ConfLongDemo_JSI.small.group.txt/part-r-00000






.. raw:: html

    <pre>
    lying	267
    falling	30
    sitting	435
    walking	170
    sitting down	56
    standing up from sitting	42

    </pre>



.. raw:: html

   <h3 id="exo2">

Exercice 2 : JOIN

.. raw:: html

   </h3>

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

.. code:: ipython3

    %%PIG solution_groupby_join.pig
    
    myinput = LOAD 'ConfLongDemo_JSI.small.example.txt' 
              using PigStorage(',') 
              AS (index:long, sequence, tag, timestamp:long, dateformat, x:double,y:double, z:double, activity) ;
    
    gr = GROUP myinput BY activity ;
    avgact = FOREACH gr GENERATE group, COUNT(myinput) ; 
    
    joined = JOIN myinput BY activity, avgact BY group ;
    
    STORE joined INTO 'ConfLongDemo_JSI.small.group.join.txt' USING PigStorage() ;

.. code:: ipython3

    %pig_submit solution_groupby_join.pig -r groupby.join.redirection






.. raw:: html

    <pre>

    </pre>



.. code:: ipython3

    %remote_cmd tail groupby.join.redirection.err






.. raw:: html

    <pre>
    2015-10-29 01:15:15,416 [main] INFO  org.apache.hadoop.conf.Configuration.deprecation - mapred.job.tracker is deprecated. Instead, use mapreduce.jobtracker.address
    2015-10-29 01:15:15,416 [main] INFO  org.apache.hadoop.conf.Configuration.deprecation - fs.default.name is deprecated. Instead, use fs.defaultFS
    2015-10-29 01:15:15,416 [main] INFO  org.apache.pig.backend.hadoop.executionengine.HExecutionEngine - Connecting to hadoop file system at: hdfs://nameservice1
    2015-10-29 01:15:17,285 [main] INFO  org.apache.pig.tools.pigstats.ScriptState - Pig features used in the script: HASH_JOIN,GROUP_BY
    2015-10-29 01:15:17,348 [main] INFO  org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer - {RULES_ENABLED=[AddForEach, ColumnMapKeyPrune, DuplicateForEachColumnRewrite, GroupByConstParallelSetter, ImplicitSplitInserter, LimitOptimizer, LoadTypeCastInserter, MergeFilter, MergeForEach, NewPartitionFilterOptimizer, PartitionFilterOptimizer, PushDownForEachFlatten, PushUpFilter, SplitFilter, StreamTypeCastInserter], RULES_DISABLED=[FilterLogicExpressionSimplifier]}
    2015-10-29 01:15:17,404 [main] INFO  org.apache.hadoop.conf.Configuration.deprecation - mapred.textoutputformat.separator is deprecated. Instead, use mapreduce.output.textoutputformat.separator
    2015-10-29 01:15:17,426 [main] ERROR org.apache.pig.tools.grunt.Grunt - ERROR 6000: 
    <file solution_groupby_join.pig, line 11, column 0> Output Location Validation Failed for: 'hdfs://nameservice1/user/xavierdupre/ConfLongDemo_JSI.small.group.join.txt More info to follow:
    Output directory hdfs://nameservice1/user/xavierdupre/ConfLongDemo_JSI.small.group.join.txt already exists
    Details at logfile: /home/xavierdupre/pig_1446077714461.log

    </pre>



.. code:: ipython3

    %remote_cmd hdfs dfs -ls ConfLongDemo_JSI.small.group.join.txt






.. raw:: html

    <pre>
    Found 2 items
    -rw-r--r--   3 xavierdupre xavierdupre          0 2015-10-29 01:13 ConfLongDemo_JSI.small.group.join.txt/_SUCCESS
    -rw-r--r--   3 xavierdupre xavierdupre     144059 2015-10-29 01:13 ConfLongDemo_JSI.small.group.join.txt/part-r-00000

    </pre>



.. code:: ipython3

    %remote_cmd hdfs dfs -tail ConfLongDemo_JSI.small.group.join.txt/part-r-00000






.. raw:: html

    <pre>
    26262834000	27.05.2009 14:03:46:283	3.3038318157196045	1.938292145729065	0.7622964978218079	standing up from sitting	standing up from sitting	42
    652	A01	020-000-033-111	633790226262563704	27.05.2009 14:03:46:257	3.2363295555114746	2.00623106956482	1.1472841501235962	standing up from sitting	standing up from sitting	42
    651	A01	010-000-030-096	633790226262293413	27.05.2009 14:03:46:230	3.275949239730835	1.7746492624282837	0.3117055296897888	standing up from sitting	standing up from sitting	42
    650	A01	010-000-024-033	633790226262023117	27.05.2009 14:03:46:203	3.2498104572296143	1.878917098045349	0.13854867219924927	standing up from sitting	standing up from sitting	42
    649	A01	020-000-032-221	633790226261752823	27.05.2009 14:03:46:177	3.352446317672729	1.950886845588684	0.8281049728393555	standing up from sitting	standing up from sitting	42
    648	A01	020-000-033-111	633790226261482530	27.05.2009 14:03:46:147	3.2220029830932617	2.0042579174041752	1.032345414161682	standing up from sitting	standing up from sitting	42

    </pre>



.. raw:: html

   <h3 id="prol">

Prolongements

.. raw:: html

   </h3>

`PIG <http://pig.apache.org/>`__ n’est pas la seule façon d’exécuter des
jobs Map/Reduce. `Hive <https://hive.apache.org/>`__ est un langage dont
la syntaxe est très proche de celle du SQL. L’article `Comparing Pig
Latin and SQL for Constructing Data Processing
Pipelines <https://developer.yahoo.com/blogs/hadoop/comparing-pig-latin-sql-constructing-data-processing-pipelines-444.html>`__
explicite les différences des deux approches.

**langage haut niveau**

Ce qu’il faut retenir est que le langage PIG est un langage haut niveau.
Le programme est compilé en une séquence d’opérations Map/Reduce
transparente pour l’utilisateur. Le temps de développement est très
réduit lorsqu’on le compare au même programme écrit en Java. Le
compilateur construit un plan d’exécution (`quelques exemples
ici <http://chimera.labs.oreilly.com/books/1234000001811/ch07.html#explain>`__)
et infère le nombre de machines requises pour distribuer le job. Cela
suffit pour la plupart des besoins, cela nécessite.

**petits jeux**

Certains jobs peuvent durer des heures, il est conseillée de les essayer
sur des petits jeux de données avant de les faire tourner sur les vrais
données. Il est toujours frustrant de s’apercevoir qu’un job a planté au
bout de deux heures car une chaîne de caractères est vide et que ce cas
n’a pas été prévu.

Avec ces petits jeux, il est possible de faire tourner et conseillé de
tester le job d’abord sur la passerelle (`exécution
local <http://archive.cloudera.com/cdh/3/pig/tutorial.html#Running+the+Pig+Scripts+in+Local+Mode>`__)
avant de le lancer sur le cluster. Avec pyensae, il faut ajouter
l’option ``-local`` à la commande
`pig_submit <http://www.xavierdupre.fr/app/pyensae/helpsphinx/pyensae/remote/magic_remote_ssh.html?highlight=pig_submit#pyensae.remote.magic_remote_ssh.MagicRemoteSSH.pig_submit>`__.

**concaténer les fichiers divisés**

Un programme PIG ne produit pas un fichier mais plusieurs fichiers dans
un répertoire. La commande
`getmerge <http://hadoop.apache.org/docs/r2.3.0/hadoop-project-dist/hadoop-common/FileSystemShell.html>`__
télécharge ces fichiers sur la passerelle et les fusionne en un seul.

**ordre des lignes**

Les jobs sont distribués, même en faisant rien (LOAD + STORE), il n’est
pas garanti que l’ordre des lignes soit préservé. La probabilié que ce
soit le cas est quasi nulle.