0317 09:53:14.865+0000 {12772} INFO  [pm-worker-exec slot-Task:id=8274,env=12772,type=11][c.s.w.t.f.s.PostExecutionStage  ] Loaded {child runId vs completion type}: {8286-SUCCESSFUL}{8287-SUCCESSFUL}{8288-SUCCESSFUL}{8289-SUCCESSFUL}{8290-SUCCESSFUL}{8291-SUCCESSFUL}{8292-SUCCESSFUL}{8293-SUCCESSFUL}{8294-SUCCESSFUL}{8295-SUCCESSFUL}{8296-SUCCESSFUL}
0317 09:54:12.498+0000 {12772} INFO  [pm-worker-exec slot-Task:id=8273,env=12772,type=55][edProcessInputBatchKafkaProducer] Started producing records on topic 12772_8286_20170317_IN_0
0317 09:54:13.428+0000 {12772} INFO  [pm-worker-exec slot-Task:id=8273,env=12772,type=55][edProcessInputBatchKafkaProducer] Started producing records on topic 12772_8287_20170317_IN_0 
0317 09:55:13.027+0000 {12772} INFO  [pm-worker-exec slot-Task:id=8273,env=12772,type=55][edProcessInputBatchKafkaProducer] Done with producing records on topic 12772_8286_20170317_IN_0
0317 09:55:15.027+0000 {12772} INFO  [pm-worker-exec slot-Task:id=8273,env=12772,type=55][edProcessInputBatchKafkaProducer] Done with producing records on topic 12772_8287_20170317_IN_0

##</code><code>##

 
topic,start_time,Endtime
12772_8286_20170317_IN_0,09:54:12.498+0000,09:55:13.027+0000
12772_8287_20170317_IN_0,09:54:13.428+0000,09:55:15.027+0000

##</code><code>##

open (MYFILE, '>data.CSV');
$LOGFILE = "worker.log";
open(LOGFILE) or die("Could not open log file.");
$LOGFILE1 = "worker.log";
open(LOGFILE1) or die("Could not open log file.");

$taskid=8274;


foreach $line (<LOGFILE>) 
{
    chomp($line); 
	if ( $line =~ m/$taskid.*child runId vs completion type/)
	{
	
	@Task_id = $line =~ /\{(\d+)-/g;
    
	}	
}


foreach $line (<LOGFILE1>) 
{
    chomp($line); 
	foreach $childid(@Task_id)
	{
	if ( $line =~ m/Started producing records on topic (\d+)_($childid)_(\d+)_IN_(\d+)/)
	  {
	  $envId=$1;
	  $timestamp=$3;
	  $input=$4;
	my @fields = split / /, $line;
	$data1="$envId\_$childid\_$timestamp\_IN\_$input\,$fields[1]";
	$data2="$envId\_$childid\_$timestamp\_IN\_$input";
	push @data9,$data1 ;
	print "$data1\n";
	  }
	  
	 if ( $line =~ m/Done with producing records on topic (\d+)_($childid)_(\d+)_IN_(\d+)/)
	  {
	  $envId=$1;
	  $timestamp=$3;
	  $input=$4;
	my @fields = split / /, $line;
	$data3="$envId\_$childid\_$timestamp\_IN\_$input\,$fields[1]";
	$data4="$envId\_$childid\_$timestamp\_IN\_$input";
	push @data9,$data3 ;
	print "$data3\n";
	  }
	 
	 }
    	  
}
foreach (@data9) {
 print MYFILE "$_\n";
}	
	
	
close LOGFILE;
close LOGFILE1;