Great question. My solution below is trimmed out of the presentation described at https://martin.atlassian.net/wiki/x/GYBzAg on a much bigger topic. So, let's assume you have an this original file that has an ID, a date created, and three "payload" attributes. [root@sandbox ~]# hdfs dfs -cat origs.csv
11,2014-09-17,base,base,base
12,2014-09-17,base,base,base
13,2014-09-17,base,base,base
14,2014-09-18,base,base,base
15,2014-09-18,base,base,base
16,2014-09-18,base,base,base
17,2014-09-19,base,base,base
18,2014-09-19,base,base,base
19,2014-09-19,base,base,base Now, let's assume you have a delta file that has 4 new records (ID's 10, 20, 21 and 22) as well as more recent versions of 3 other records (IDs 12, 14, and 16). [root@sandbox ~]# hdfs dfs -cat delta.csv
10,2014-09-16,oops,was,missed
20,2014-09-20,base,base,base
21,2014-09-20,base,base,base
22,2014-09-20,base,base,base
12,2014-09-17,base,CHANGED,base
14,2014-09-18,base,CHANGED,base
16,2014-09-18,base,CHANGED,base Then, in a Pig script you could join these like this. origs = LOAD '/user/maria_dev/hcc/86778/original.csv'
USING PigStorage(',') AS
( bogus_id:int, date_cr: chararray,
field1:chararray, field2:chararray, field3:chararray );
delta = LOAD '/user/maria_dev/hcc/86778/delta.csv'
USING PigStorage(',') AS
( bogus_id:int, date_cr: chararray,
field1:chararray, field2:chararray, field3:chararray );
joined = JOIN origs BY bogus_id FULL OUTER, delta BY bogus_id;
DESCRIBE joined;
DUMP joined; And get this output. joined: {origs::bogus_id: int,origs::date_cr: chararray,origs::field1: chararray,origs::field2: chararray,origs::field3: chararray,delta::bogus_id: int,delta::date_cr: chararray,delta::field1: chararray,delta::field2: chararray,delta::field3: chararray}
(,,,,,10,2014-09-16,oops,was,missed)
(11,2014-09-17,base,base,base,,,,,)
(12,2014-09-17,base,base,base,12,2014-09-17,base,CHANGED,base)
(13,2014-09-17,base,base,base,,,,,)
(14,2014-09-18,base,base,base,14,2014-09-18,base,CHANGED,base)
(15,2014-09-18,base,base,base,,,,,)
(16,2014-09-18,base,base,base,16,2014-09-18,base,CHANGED,base)
(17,2014-09-19,base,base,base,,,,,)
(18,2014-09-19,base,base,base,,,,,)
(19,2014-09-19,base,base,base,,,,,)
(,,,,,20,2014-09-20,base,base,base)
(,,,,,21,2014-09-20,base,base,base)
(,,,,,22,2014-09-20,base,base,base) You'll see above that if the delta record's fields are present (the ones on the right side) then they should be the ones carried forward as they are either new (4) or modified (3) records, but if they are missing (6) then the original values should just roll forward. merged = FOREACH joined GENERATE
((delta::bogus_id is not null) ? delta::bogus_id: origs::bogus_id) as bogus_id,
((delta::date_cr is not null) ? delta::date_cr: origs::date_cr) as date_cr,
((delta::field1 is not null) ? delta::field1: origs::field1) as field1,
((delta::field2 is not null) ? delta::field2: origs::field2) as field2,
((delta::field3 is not null) ? delta::field3: origs::field3) as field3;
DESCRIBE merged;
DUMP merged; As you can see from the combined output, we have the necessary 13 rows in the new dataset. merged: {bogus_id: int,date_cr: chararray,field1: chararray,field2: chararray,field3: chararray}
(10,2014-09-16,oops,was,missed)
(11,2014-09-17,base,base,base)
(12,2014-09-17,base,CHANGED,base)
(13,2014-09-17,base,base,base)
(14,2014-09-18,base,CHANGED,base)
(15,2014-09-18,base,base,base)
(16,2014-09-18,base,CHANGED,base)
(17,2014-09-19,base,base,base)
(18,2014-09-19,base,base,base)
(19,2014-09-19,base,base,base)
(20,2014-09-20,base,base,base)
(21,2014-09-20,base,base,base)
(22,2014-09-20,base,base,base) Good luck and happy Hadooping!
... View more
