collectl_parser.pig

%default HARDWARE 'pi0w'
%default LOADPATH '/storage/ssd01/ES/data/$HARDWARE/'
%default ESINDEX 'benchmark/$HARDWARE'

register lib/httpclient-4.5.3.jar;
register contrib/piggybank/java/piggybank.jar;
register elasticsearch-hadoop-5.4.3.jar;
DEFINE EsStorage org.elasticsearch.hadoop.pig.EsStorage('localhost');

cpu_data = LOAD '$LOADPATH/cpu/*'  USING org.apache.pig.piggybank.storage.CSVExcelStorage(' ', 'NO_MULTILINE', 'UNIX', 'READ_INPUT_HEADER') as (date:chararray, time:chararray, cpu_user:int, cpu_nice:int, cpu_sys:int, cpu_wait:int, cpu_irq:int, cpu_soft:int, cpu_steal:int, cpu_idle:int, cpu_total:int, cpu_guest:int, cpu_guestn:int, cpu_intrpt:int,  cpu2_user:int, cpu2_nice:int, cpu2_sys:int, cpu2_wait:int, cpu2_irq:int, cpu2_soft:int, cpu2_steal:int, cpu2_idle:int, cpu2_total:int, cpu2_guest:int, cpu2_guestn:int, cpu2_intrpt:int,  cpu3_user:int, cpu3_nice:int, cpu3_sys:int, cpu3_wait:int, cpu3_irq:int, cpu3_soft:int, cpu3_steal:int, cpu3_idle:int, cpu3_total:int, cpu3_guest:int, cpu3_guestn:int, cpu3_intrpt:int, cpu4_user:int, cpu4_nice:int, cpu_sys4:int, cpu4_wait:int, cpu4_irq:int, cpu4_soft:int, cpu4_steal:int, cpu4_idle:int, cpu4_total:int, cpu4_guest:int, cpu4_guestn:int, cpu4_intrpt:int);
cpu_clean = FILTER cpu_data BY NOT( $0 MATCHES '.*#.*');
cpu_date = foreach cpu_clean generate ToDate(CONCAT(date,' ',time,'Z'), 'yyyyMMdd HH:mm:ssZ') as timstim, *, '$HARDWARE' as entry;
STORE cpu_date INTO '$ESINDEX' USING org.elasticsearch.hadoop.pig.EsStorage
             ('es.http.timeout = 5m',
              'es.index.auto.create = true',
              'es.mapping.names=timstim:@timestamp' );




disk_data = LOAD '$LOADPATH/disk/*'  USING org.apache.pig.piggybank.storage.CSVExcelStorage(' ', 'NO_MULTILINE', 'UNIX', 'READ_INPUT_HEADER') as (date:chararray, time:chararray, name:chararray, reads:int, rmerge:int, rkbytes:int, writes:int, wmerge:int, wkbytes:int, request:int, quelen:int, wait:int, scvtim:int, util:int);
disk_clean = FILTER disk_data BY NOT( $0 MATCHES '.*#.*');
disk_date = FOREACH disk_clean generate ToDate(CONCAT(date,' ',time,'Z'), 'yyyyMMdd HH:mm:ssZ') as timstim, *, '$HARDWARE' as entry;
STORE disk_date INTO '$ESINDEX' USING org.elasticsearch.hadoop.pig.EsStorage
             ('es.http.timeout = 5m',
              'es.index.auto.create = true',
              'es.mapping.names=timstim:@timestamp' );




mem_data = LOAD '$LOADPATH/mem/*'  USING org.apache.pig.piggybank.storage.CSVExcelStorage(' ', 'NO_MULTILINE', 'UNIX', 'READ_INPUT_HEADER') as (date:chararray, time:chararray, total:int, used:int, free:int, shared:int, buffer:int, cached:int, slab:int, map:int, anon:int, commit:int, locked:int, swaptot:int, swapused:int, swapfree:int, swapin:int, swapout:int, dirty:int, clean:int, laundry:int, inactive:int, pagein:int, pageout:int, pagefaults:int, pagemajfaults:int, hugetotal:int, hugefree:int, hugereserved:int, sunreclaim:int);
mem_clean = FILTER mem_data BY NOT( $0 MATCHES '.*#.*');
mem_date = FOREACH mem_clean generate ToDate(CONCAT(date,' ',time,'Z'), 'yyyyMMdd HH:mm:ssZ') as timstim, *, '$HARDWARE' as entry;
STORE mem_date INTO '$ESINDEX' USING org.elasticsearch.hadoop.pig.EsStorage
             ('es.http.timeout = 5m',
              'es.index.auto.create = true',
              'es.mapping.names=timstim:@timestamp' );



net_data = LOAD '$LOADPATH/net/*'  USING org.apache.pig.piggybank.storage.CSVExcelStorage(' ', 'NO_MULTILINE', 'UNIX', 'READ_INPUT_HEADER') as (date:chararray, time:chararray, name1:chararray, rxpacket1:int, txpacket1:int, rxkb1:int, txkb1:int, rxerror1:int, rxdrop1:int, rxfifo1:int, rxfra1:int, rxcmp1:int, rxmlt1:int, txerror1:int, name2:chararray, rxpacket2:int, txpacket2:int, rxkb2:int, txkb2:int, rxerror2:int, rxdrop2:int, rxfifo2:int, rxfra2:int, rxcmp2:int, rxmlt2:int, txerror2:int, name:chararray, rxpacket3:int, txpacket3:int, rxkb3:int, txkb3:int, rxerror3:int, rxdrop3:int, rxfifo3:int, rxfra3:int, rxcmp3:int, rxmlt3:int, txerror3:int);
net_clean = FILTER net_data BY NOT( $0 MATCHES '.*#.*');
net_date = foreach net_clean generate ToDate(CONCAT(date,' ',time,'Z'), 'yyyyMMdd HH:mm:ssZ') as timstim, *, '$HARDWARE' as entry;
STORE net_date INTO '$ESINDEX' USING org.elasticsearch.hadoop.pig.EsStorage
             ('es.http.timeout = 5m',
              'es.index.auto.create = true',
              'es.mapping.names=timstim:@timestamp' );
/*
#Date Time [NET:lo]Name [NET:lo]RxPkt [NET:lo]TxPkt [NET:lo]RxKB [NET:lo]TxKB [NET:lo]RxErr [NET:lo]RxDrp [NET:lo]RxFifo [NET:lo]RxFra [NET:lo]RxCmp [NET:lo]RxMlt [NET:lo]TxErr [NET:lo]TxDrp [NET:lo]TxFifo [NET:lo]TxColl [NET:lo]TxCar [NET:lo]TxCmp [NET:lo]RxErrs [NET:lo]TxErrs [NET:wwan0]Name [NET:wwan0]RxPkt [NET:wwan0]TxPkt [NET:wwan0]RxKB [NET:wwan0]TxKB [NET:wwan0]RxErr [NET:wwan0]RxDrp [NET:wwan0]RxFifo [NET:wwan0]RxFra [NET:wwan0]RxCmp [NET:wwan0]RxMlt [NET:wwan0]TxErr [NET:wwan0]TxDrp [NET:wwan0]TxFifo [NET:wwan0]TxColl [NET:wwan0]TxCar [NET:wwan0]TxCmp [NET:wwan0]RxErrs [NET:wwan0]TxErrs [NET:wlan0]Name [NET:wlan0]RxPkt [NET:wlan0]TxPkt [NET:wlan0]RxKB [NET:wlan0]TxKB [NET:wlan0]RxErr [NET:wlan0]RxDrp [NET:wlan0]RxFifo [NET:wlan0]RxFra [NET:wlan0]RxCmp [NET:wlan0]RxMlt [NET:wlan0]TxErr [NET:wlan0]TxDrp [NET:wlan0]TxFifo [NET:wlan0]TxColl [NET:wlan0]TxCar [NET:wlan0]TxCmp [NET:wlan0]RxErrs [NET:wlan0]TxErrs


#Date Time [MEM]Tot [MEM]Used [MEM]Free [MEM]Shared [MEM]Buf [MEM]Cached [MEM]Slab [MEM]Map [MEM]Anon [MEM]Commit [MEM]Locked [MEM]SwapTot [MEM]SwapUsed [MEM]SwapFree [MEM]SwapIn [MEM]SwapOut [MEM]Dirty [MEM]Clean [MEM]Laundry [MEM]Inactive [MEM]PageIn [MEM]PageOut [MEM]PageFaults [MEM]PageMajFaults [MEM]HugeTotal [MEM]HugeFree [MEM]HugeRsvd [MEM]SUnreclaim

#Date Time [DSK:mmcblk0]Name [DSK:mmcblk0]Reads [DSK:mmcblk0]RMerge [DSK:mmcblk0]RKBytes [DSK:mmcblk0]Writes [DSK:mmcblk0]WMerge [DSK:mmcblk0]WKBytes [DSK:mmcblk0]Request [DSK:mmcblk0]QueLen [DSK:mmcblk0]Wait [DSK:mmcblk0]SvcTim [DSK:mmcblk0]Util [DSK:sda]Name [DSK:sda]Reads [DSK:sda]RMerge [DSK:sda]RKBytes [DSK:sda]Writes [DSK:sda]WMerge [DSK:sda]WKBytes [DSK:sda]Request [DSK:sda]QueLen [DSK:sda]Wait [DSK:sda]SvcTim [DSK:sda]Util

*/