How to flatten array of Tuple

[java8964 <ja...@hotmail.com>]

Hi, 
For PIG 12.0, I have the following data stored in the AVRO format:
grunt> describe all_contacts;all_contacts: {account_id: long,contact_id: chararray,sequence_id: int,state: chararray,name: chararray,kind: chararray,prefix_name: chararray,first_name: chararray,middle_name: chararray,company_name: chararray,job_title: chararray,source_name: chararray,source_details: chararray,provider_name: chararray,provider_details: chararray,created_at: long,create_source: chararray,updated_at: long,update_source: chararray,accessed_at: long,deleted_at: long,delta: boolean,contact_fields: {ARRAY_ELEM: (custom_field_id: chararray,value: chararray,date_value: long)},related_contacts: {ARRAY_ELEM: (related_contact_id: chararray,kind: chararray,role: chararray,created_at: long,updated_at: long,deleted_at: long)},contact_channels: {ARRAY_ELEM: (channel_id: chararray,address: chararray,state: chararray,create_source: chararray,update_source: chararray,created_at: long,updated_at: long,deleted_at: long,description: chararray,history: chararray,source_ip: long,optin_at: long,optout_at: long,confirmed: chararray)},contact_notes: {ARRAY_ELEM: (note_id: chararray,content: chararray,created_at: long,updated_at: long)},contact_service_addresses: {ARRAY_ELEM: (service_address_id: chararray,type: chararray,address: chararray,sort_order: int,kind: chararray,created_at: long,updated_at: long,create_source: chararray,update_source: chararray,deleted_at: long)},contact_street_addresses: {ARRAY_ELEM: (street_address_id: chararray,kind: chararray,street: chararray,city: chararray,state: chararray,postal_code: chararray,country: chararray,sort_order: int,created_at: long,create_source: chararray,updated_at: long,update_source: chararray,deleted_at: long)}}
As you can see, for each row of contact, there is an array of associated channels, which itself is struct in AVRO, or Tuple in Pig. Same apply to contact_service_address.And you can see there are "address" field in both contact_channels and contact_service_addresses tuples. Now my requirement is to compare the address from contact_channels to contact_service_addresses, if matched, count this channel as one; otherwise, ignore this channel.
I tried different way in PIG for this request, but now I kind of facing some problems. The big one is how to keep the Tuple together, and only compare one subfield to another tuple's subfield.
I am using the following example to describe what I did, and what is my current problem.
-- load all the contactsall_contacts = LOAD 'data' USING org.apache.pig.piggybank.storage.avro.AvroStorage();
-- generate contact, channels and service_addresses with needed fieldsnon_deleted_channels_contacts = FOREACH all_contacts {    csa_addresses = FOREACH contact_service_addresses GENERATE address;    channels = FOREACH contact_channels GENERATE channel_id, address, deleted_at, state;    non_deleted_channels = FILTER channels BY state != 'D';    GENERATE account_id, contact_id, deleted_at as contact_deleted_at, csa_addresses, non_deleted_channels;};
-- now I kind of not sure how to do it, here is what I tried:match_address_non_deleted_channels_contacts = FOREACH non_deleted_channels_contacts     GENERATE account_id, contact_id, contact_deleted_at, FLATTEN(csa_addresses.address) as csa_address, FLATTEN(non_deleted_channels.address) as cc_address,    FLATTEN(non_deleted_channels.channel_id) as cc_channel_id;
I don't want to flatten the data, but I cannot do it in nested FOREACH, and FOREACH can only do 2 levels nested in PIG. But if I want to compare the address of channel to the address of service_address, I have to use 3 level FOREACH, as following:FOREACH contacts {    FOREACH service_addresses {        FOREACH channels {        }    }}
which is not supported in PIG 12. So I gave up this way.
But if I use FLATTEN, I have to FLATTEN both (channels.address) and (channels.channel_id), which expands the data not the way I want.For example, if one contact has 4 service_address, and 2 channels, what I assume after FLATTEN it will be 8 rows for this contact, as (4 x 2 = 8). But what pig gives me back is 16 rows, as (4 service_address.address x 2 channels.address x 2 channels.channel_id = 16 rows).The pig treats the address and channel_id from Channel tuple as separated elements in the flatten operation, which makes the rows grow way too fast. If I need additional fields out from the channel tuple, it just explode too much unnecessary. But I don't know how to flatten the array of tuple together, as right now I can only flatten one field of tuple individually.
If I do this:
match_address_non_deleted_channels_contacts = FOREACH non_deleted_channels_contacts     GENERATE account_id, contact_id, contact_deleted_at, FLATTEN(csa_addresses.address) as csa_address, FLATTEN(non_deleted_channels) as channels;
It will give me error:2014-04-17 15:02:16,384 [main] ERROR org.apache.pig.tools.grunt.Grunt - ERROR 1031: Incompatable schema: left is "channels:NULL", right is "non_deleted_channels::channel_id:chararray,non_deleted_channels::address:chararray,non_deleted_channels::deleted_at:long,non_deleted_channels::state:chararray"
Is there a way to flatten the array of tuple out as a whole tuple, so I can use individual fields later, instead of flattening individual fields out, make the rows explode wrong?
I google this case, only suggestion I can find looks like is to NOT flatten the array, but then in my case, I cannot compare the field from one tuple to another field in another tuple.
I hope I explain my problem clearly. Any suggestion is appreciated.
Thanks
Yong