Traiter les données avec Pig

-- Charger les données

truck_events = LOAD 'truck_event_text_partition.csv' USING PigStorage(',')
AS (driverId:int, truckId:int, eventTime:chararray,
	eventType:chararray, longitude:double, latitude:double,
	eventKey:chararray, correlationId:long, driverName:chararray,
	routeId:long,routeName:chararray,eventDate:chararray);

drivers =  LOAD 'drivers.csv' USING PigStorage(',')
	AS (driverId:int, name:chararray, ssn:chararray,
		location:chararray, certified:chararray, wage_plan:chararray);

timesheet = LOAD 'timesheet.csv' USING PigStorage(',')
AS (driverId:int,week:int,hours:int,miles:int);

-- Filtrer pour enlever la première ligne

truck_events = filter truck_events by driverId > 0;
drivers  = filter drivers by driverId > 0;
timesheet = filter timesheet by driverId > 0;


-- Juste les donnees qui nous interessent

chauffeurs = FOREACH drivers GENERATE driverId, name, location ;
evenements = FOREACH truck_events GENERATE driverId, eventType ;

-- compter les heures et les miles

gpd = GROUP timesheet BY driverId;
sommes = FOREACH gpd GENERATE
			group AS driverId,
			SUM(timesheet.hours) AS hours,
			SUM(timesheet.miles) AS miles;

-- Recenser les infractions

infractions =  FILTER evenements BY NOT (eventType MATCHES 'Normal');
infractions_groupees = GROUP infractions BY driverId;
nb_infractions = FOREACH infractions_groupees GENERATE
				group AS driverId,
				COUNT(infractions) as nb;

-- jointures

j = JOIN chauffeurs BY driverId, sommes BY driverId ;
jj = JOIN j BY sommes::driverId LEFT OUTER, nb_infractions BY driverId;

jjj = FOREACH jj generate
		j::chauffeurs::name AS nom,
		j::chauffeurs::location AS adresse,
		j::sommes::hours AS hours,
		j::sommes::miles AS miles,
		nb_infractions::nb AS nb ;

-- sortie finale
resultat = ORDER jjj by nom;

STORE resultat INTO 'statistiques_chauffeurs' USING PigStorage(',');

Arcs = LOAD 'reseau.txt'  USING PigStorage('\t') AS (frm:int,to:int);

GrSort = GROUP Arcs BY frm;

N = FOREACH (group GrSort ALL)
    GENERATE COUNT(GrSort);

deg_sort = FOREACH GrSort
           GENERATE  group AS ID, COUNT(Arcs) AS degs ;


iter0 = FOREACH deg_sort GENERATE ID, (float) 1 / (float)N.$0 AS PR0, (float) 1 / (float)N.$0 AS PR, degs ;

-- une première itération

contribs = FOREACH iter0 GENERATE ID, PR / degs AS contrib;
Arcscontribs = JOIN Arcs BY frm, contribs BY ID ;
contribsParBut = GROUP Arcscontribs BY to;
sommesContribs = FOREACH contribsParBut GENERATE group AS ID, SUM(Arcscontribs.contrib) AS PR ;
iterSommesContribs = JOIN iter0 BY ID, sommesContribs BY ID;
iter1 = FOREACH iterSommesContribs GENERATE iter0::ID AS ID, iter0::PR0 AS PR0, 0.15 * iter0::PR0 + 0.85 * sommesContribs::PR AS PR, iter0::degs AS degs;

DEFINE iteration(precedent,A) RETURNS suivant {
	contribs = FOREACH $precedent GENERATE ID, PR / degs AS contrib;
	ArcsContribs = JOIN $A BY frm, contribs BY ID ;
	contribsParBut = GROUP ArcsContribs BY to;
	sommesContribs = FOREACH contribsParBut GENERATE group AS ID, SUM(ArcsContribs.contrib) AS PR ;
	iterSommesContribs = JOIN $precedent BY ID, sommesContribs BY ID;
	$suivant = FOREACH iterSommesContribs GENERATE $precedent::ID AS ID, $precedent::PR0 AS PR0, 0.15 * $precedent::PR0 + 0.85 * sommesContribs::PR AS PR, $precedent::degs AS degs;
}

Arcs = LOAD 'reseau.txt'  USING PigStorage('\t') AS (frm:int,to:int);

GrSort = GROUP Arcs BY frm;

N = FOREACH (group GrSort ALL)
    GENERATE COUNT(GrSort);

deg_sort = FOREACH GrSort
           GENERATE  group AS ID, COUNT(Arcs) AS degs ;


iter0 = FOREACH deg_sort GENERATE ID, (float) 1 / (float)N.$0 AS PR0, (float) 1 / (float)N.$0 AS PR, degs ;

DEFINE iteration(precedent,A) RETURNS suivant {
	contribs = FOREACH $precedent GENERATE ID, PR / degs AS contrib;
	ArcsContribs = JOIN $A BY frm, contribs BY ID ;
	contribsParBut = GROUP ArcsContribs BY to;
	sommesContribs = FOREACH contribsParBut GENERATE group AS ID, SUM(ArcsContribs.contrib) AS PR ;
	iterSommesContribs = JOIN $precedent BY ID, sommesContribs BY ID;
	$suivant = FOREACH iterSommesContribs GENERATE $precedent::ID AS ID, $precedent::PR0 AS PR0, 0.15 * $precedent::PR0 + 0.85 * sommesContribs::PR AS PR, $precedent::degs AS degs;
}

iter1 = iteration(iter0,Arcs);
iter2 = iteration(iter1,Arcs);
iter3 = iteration(iter2,Arcs);
iter4 = iteration(iter3,Arcs);
iter5 = iteration(iter4,Arcs);
iter6 = iteration(iter5,Arcs);
iter7 = iteration(iter6,Arcs);
iter8 = iteration(iter7,Arcs);
iter9 = iteration(iter8,Arcs);


pageranks = FOREACH iter9 GENERATE ID, PR;

STORE pageranks INTO 'resultat' USING PigStorage('\t') ;