<jupyter_start><jupyter_text>PIG et JSON et streaming avec les données vélib - correction avec clouderaCorrection.<jupyter_code>from jyquickhelper import add_notebook_menu
add_notebook_menu()<jupyter_output><empty_output><jupyter_text>Récupération des donnéesLe 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).<jupyter_code>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]<jupyter_output><empty_output><jupyter_text>Connexion au cluster<jupyter_code>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")
import pyensae
%load_ext pyensae
password = params["password"]
server = params["server"]
username = params["username"]
client = %remote_open
client<jupyter_output><empty_output><jupyter_text>Exercice 1 : convertir les valeurs numériquesLe programme suivant prend comme argument les colonnes à extraire des fichiers textes qui sont enregistrés au format "python".<jupyter_code>%%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()<jupyter_output><empty_output><jupyter_text>On vérifie qu'il fonctionne en lui faisant ingérer les deux lignes suivantes :<jupyter_code>%%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}]<jupyter_output><empty_output><jupyter_text>On écrit le script PIG qui utilise plus de colonnes :<jupyter_code>%%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') ;<jupyter_output><empty_output><jupyter_text>On supprime la précédente exécution si besoin puis on vérifie que le répertoire contenant les résultats est vide :<jupyter_code>if client.dfs_exists("velib_py_results/firstjob"):
    client.dfs_rm("velib_py_results/firstjob", recursive=True)
%dfs_mkdir velib_py_results
%dfs_ls velib_py_results/
%pig_submit json_velib.pig stream_json.py -r redirection
%remote_cmd tail redirection.pig.err<jupyter_output><empty_output><jupyter_text>On récupère les informations qu'on affiche sous forme de dataframe :<jupyter_code>if os.path.exists("velib_exo1.txt") : os.remove("velib_exo1.txt")
client.download_cluster("velib_py_results/firstjob","velib_exo1.txt", merge=True)
import pandas
df = pandas.read_csv("velib_hd.txt", sep="\t",names=["available_bike_stands","available_bikes","lat","lng","name","status"])
df.head()<jupyter_output><empty_output><jupyter_text>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 :<jupyter_code>%dfs_rmr velib_py_results/fermees.txt
%dfs_rmr velib_py_results/distribution_bikes.txt
%dfs_rmr velib_py_results/distribution_stands.txt
%%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') ;
%pig_submit json_fermee.pig stream_json.py -r redirection
%remote_cmd tail redirection.pig.err 
%dfs_ls velib_py_results
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
import pandas
df = pandas.read_csv("distribution_bikes.txt", sep="\t", names=["nb_velos", "nb_stations_minutes"])
df.head()
df.plot(x="nb_velos",y="nb_stations_minutes",kind="bar",figsize=(16,4))
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
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))
if os.path.exists("fermees.txt") : os.remove("fermees.txt")
%remote_down_cluster_merge velib_py_results/fermees.txt fermees.txt
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<jupyter_output><empty_output><jupyter_text>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).<jupyter_code>%dfs_ls /user/xavierdupre/
%%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') ;
%pigsubmit json_fermee.pig stream_json.py redirection
%remote_cmd tail redirection.err
%dfs_ls velib_py_results_3days/*<jupyter_output><empty_output><jupyter_text>On voit que le job a été distribué sur trois machines.  Exercice 4 : astucesLes 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<jupyter_code>%%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()
%%runpy stream_json_err.py
n'importe quoi
%%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' ;
%pig_submit test.pig stream_json_err.py -r redirection
%remote_cmd tail redirection.pig.err -n 5
if os.path.exists("python_info2.txt") : os.remove("python_info2.txt")
%remote_down_cluster_merge python_info2.txt python_info2.txt
%head python_info2.txt<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1106636315881632, 'fraction_numerical': 0.02039922955699527, 'mean_word_length': 4.725814536340852, 'pattern_counts': {'":': 3, '<': 45, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '469403', 'license': 'MIT', 'n_tokens_mistral': 4234, 'n_tokens_neox': 3958, 'n_words': 958, 'path': '_doc/notebooks/pig_hive/pig_streaming_cloudera_correction.ipynb', 'repo_name': 'sdpython/sparkouille', 'size': '11422'}