How to denormalized 2 csv files with cardinality 0,1 and mainly 1,n with Pig?

Question

I need some help for my pig script. I have 2 csv file and I want to do a join between them with a common id.

customer.csv :
1   ; nom1   ; prenom1   
2   ; nom2   ; prenom2   
3   ; nom3   ; prenom3   


child.csv
1  ; enfant_1_1  
2  ; enfant_1_2  
3  ; enfant_1_3  
1  ; enfant_2_1  
1  ; enfant_3_1

So one customer could have many child but a child could have only one "customer".

I want to create this file :

1   ; nom1   ; prenom1  ; enfant_1_1  ; enfant_2_1  ; enfant_3_1    
2   ; nom2   ; prenom2  ; enfant_1_2   
3   ; nom3   ; prenom3  ; enfant_1_3

This is my method :

First I try do have :

1  ; enfant_1_1  ; enfant_2_1  ; enfant_3_1
2  ; enfant_1_2
3  ; enfant_1_3

And after I will do the join with custome.csv

Tell me I you think there are an easiest way :)

This is my script :

donnees_Enfants = LOAD '/user/cloudera/Jeux/mini_jeu2.csv' USING PigStorage(';')
AS (id_parent:int,nom_enfant:chararray);

group_enfants = GROUP donnees_Enfants BY id_parent;

enfant_uneLigne = foreach group_enfants generate group, donnees_Enfants.nom_enfant;

grunt> echantillon = LIMIT enfant_uneLigne 50;
grunt> DUMP echantillon;

With the DESCRIBE : group_enfants: {group: int,donnees_Enfants: {(id_parent: int,nom_enfant: chararray)}} enfant_uneLigne: {group: int,{(nom_enfant: chararray)}}

The result :

(1,{( enfant_2_1  ),( enfant_1_1  ),( enfant_3_1  )})
(2,{( enfant_2_2  )})
(3,{( enfant_2_3  )})

I tried to flatten "enfant_1_2" ... but the consequences was to had a lign per child... I have some difficulties to play with the tuple and the bags, can you help me ?

Thanks in advance,

Edit : I found a solution to my problem and more ^^ see below

Angelik

Angelik Angelik · Accepted Answer · 2014-03-25T14:33:05

Finally, I found the solution and it works with more fields for child : (id, name, age).

-- 1. Load the two files

donnees_Enfants = LOAD '/user/cloudera/JeuxDenormalisation/Jeux/mini_jeu2.csv' USING PigStorage(';') AS (id:int,nom_enfant:chararray);

donnees_Parents = LOAD '/user/cloudera/JeuxDenormalisation/Jeux/mini_jeu1.csv' USING PigStorage(';') AS (id_parent:int,nom_parent:chararray,prenom_parent:chararray);

-- 2. Join the files with the LEFT OUTER to keep customers which don't have child.

denormalisation = JOIN donnees_Parents BY id_parent LEFT OUTER, donnees_Enfants BY id ;

(9, nom9   , prenom9   ,9, enfant_2_9  )
(9, nom9   , prenom9   ,9, enfant_3_9  )
(9, nom9   , prenom9   ,9, enfant_1_9  )
(10, nom10  , prenom10  ,10, enfant_3_10)
(10, nom10  , prenom10  ,10, enfant_1_10 )
(10, nom10  , prenom10  ,10, enfant_2_10 )

-- 3. GroupBy on the customer to have only one row by customer

unParent_parLigne = GROUP denormalisation by (id_parent, nom_parent, prenom_parent);

((48, nom48  , prenom48  ),{(48, nom48  , prenom48  ,48, enfant_2_48 ),(48, nom48  , prenom48  ,48, enfant_1_48 )})
((49, nom49  , prenom49  ),{(49, nom49  , prenom49  ,49, enfant_2_49 ),(49, nom49  , prenom49  ,49, enfant_1_49 )})
((50, nom50  , prenom50  ),{(50, nom50  , prenom50  ,50, enfant_2_50 ),(50, nom50  , prenom50  ,50, enfant_1_50 )})
((51, nom51  , prenom51  ),{(51, nom51  , prenom51  ,51, enfant_1_51 )})

-- 4. FLATTEN on the rows :

ligne_finale = foreach unParent_parLigne generate FLATTEN (group), FLATTEN(BagToTuple(denormalisation.(donnees_Enfants::nom_enfant,donnees_Enfants::age)));

(9, nom9   , prenom9   , enfant_2_9  , enfant_3_9  , enfant_1_9  )
(10, nom10  , prenom10  , enfant_3_10, enfant_1_10 , enfant_2_10 )
(11, nom11  , prenom11  , enfant_1_11 , enfant_2_11 )

or if there are more fields (with "donnees_Enfants::age"):

(8, nom8   , prenom8   , enfant_3_8  , age_3_8 , enfant_2_8  , age_2_8 , enfant_1_8  , age_1_8 )
(9, nom9   , prenom9   , enfant_2_9  , age_2_9 , enfant_3_9  , age_3_9 , enfant_1_9  , age_1_9 )
(10, nom10  , prenom10  , enfant_3_10 , age_3_10, enfant_1_10 , age_1_10, enfant_2_10 , age_2_10)

-- 5. Store the data in a csv file STORE ligne_finale INTO '/user/cloudera/JeuxDenormalisation/Resultats/test4' USING org.apache.pig.piggybank.storage.PigStorageSchema(";");

How to denormalized 2 csv files with cardinality 0,1 and mainly 1,n with Pig?

1 Answers