{
task = LineParsing.tryParseTask(line);
}
}
LogData data = new LogData();
String jobId = null;
data.setCluster(_clusterName);
try
{
if (job != null)
{
// log lines are sometimes truncated, so data may be missing - just ignore these
if (job.getJobId() != null)
{
jobId = job.getJobId().toString();
job.setTasks(new ArrayList<Task>());
data.setEntry(job);
data.setPath(findInputSplitForJob(jobId,inputSplits));
context.write(new AvroKey<String>(jobId), new AvroValue<LogData>(data));
}
}
else if (attempt != null)
{
// log lines are sometimes truncated, so data may be missing - just ignore these
if (attempt.getJobId() != null && attempt.getTaskId() != null && attempt.getTaskAttemptId() != null && attempt.getType() != null)
{
jobId = attempt.getJobId().toString();
data.setEntry(attempt);
data.setPath(findInputSplitForJob(jobId,inputSplits));
context.write(new AvroKey<String>(jobId), new AvroValue<LogData>(data));
}
}
else if (task != null)
{
// log lines are sometimes truncated, so data may be missing - just ignore these
if (task.getJobId() != null && task.getTaskId() != null && task.getType() != null)
{
jobId = task.getJobId().toString();
task.setAttempts(new ArrayList<Attempt>());
data.setEntry(task);
data.setPath(findInputSplitForJob(jobId,inputSplits));
context.write(new AvroKey<String>(jobId), new AvroValue<LogData>(data));
}
}
}
catch (Exception e)
{
System.out.println("Exception writing log data: " + e.toString());
if (jobId != null)
{
System.out.println("jobId: " + jobId);
CharSequence path = data.getPath();
if (path != null)
{
System.out.println("input: " + path);
}
}