TICKET = LOAD 'cfs:///tld/TICKET_20110709.txt' USING PigStorage(',') AS (ticket_sk:long, store_sk: int, sales_type:int, store_id: int, sales_date:chararray, ticket_num, gross_sales, net_sales);
DETAIL = LOAD 'cfs:///tld/TICKET_DETAIL_20110709.txt' USING PigStorage(',') AS (ticket_detail_sk:long, store_sk:int, product_sk:long, ticket_sk:long, subheader_sk:long, store_id:int, product:int, pos_key_num, sales_date:chararray, ticket_num, sub_ticket_num, sales_type_id, product_type_id, sequence_num, condiment_sequence_num, quantity:int, item_price, actual_amount, retail_price);

TICKET_DATA = FILTER TICKET BY (ticket_sk IS NOT NULL);
DETAIL_DATA = FILTER DETAIL BY (ticket_detail_sk IS NOT NULL);

TICKET_INFO = FOREACH TICKET_DATA GENERATE ticket_sk, store_id, sales_date, sales_type, ticket_num;
DETAIL_INFO = FOREACH DETAIL_DATA GENERATE ticket_sk, store_id, sales_date, ticket_num, product, quantity, actual_amount;

TLD = GROUP TICKET_INFO BY (ticket_sk, sales_date, store_id, ticket_num), DETAIL_INFO BY (ticket_sk, sales_date, store_id, ticket_num);

TLD_ROWS = FOREACH TLD GENERATE
  CONCAT(
	CONCAT(
		CONCAT(
			CONCAT(
				REPLACE(
					SUBSTRING(group.sales_date, 0, 10),
					'-',
					'/'),
				'/'),
			(chararray) group.store_id),
		'/'),
	(chararray) group.ticket_num),
  TICKET_INFO,
  DETAIL_INFO;

STORE TLD_ROWS INTO 'cassandra://TLD/Ticket' USING CassandraStorage();