@Test(expected=ProcessingException.class) public void testCreateJobHistoryFileParserNullConf() throws IOException { final String JOB_HISTORY_FILE_NAME = "src/test/resources/job_1329348432655_0001-1329348443227-user-Sleep+job-1329348468601-10-1-SUCCEEDED-default.jhist"; File jobHistoryfile = new File(JOB_HISTORY_FILE_NAME); byte[] contents = Files.toByteArray(jobHistoryfile); JobHistoryFileParser historyFileParser = JobHistoryFileParserFactory.createJobHistoryFileParser(contents, null); assertNotNull(historyFileParser); assertTrue(historyFileParser instanceof JobHistoryFileParserHadoop2); JobKey jobKey = new JobKey("cluster1", "user", "Sleep", 1, "job_1329348432655_0001"); historyFileParser.parse(contents, jobKey); }
@Override public void parse(byte[] historyFileContents, JobKey jobKey) throws ProcessingException { this.jobKey = jobKey; this.jobKeyBytes = jobKeyConv.toBytes(jobKey); this.jobDetails = new JobDetails(jobKey); initializeJobDetails(); setJobId(jobKey.getJobId().getJobIdString()); try { FSDataInputStream in = new FSDataInputStream(new ByteArrayWrapper(historyFileContents)); String versionIgnore = in.readLine(); this.schema = schema.parse(in.readLine()); understandSchema(schema.toString()); this.reader = new GenericDatumReader<GenericRecord>(schema); this.decoder = DecoderFactory.get().jsonDecoder(schema, in); GenericRecord record = null; Hadoop2RecordType recType = null; try { while ((record = reader.read(null, decoder)) != null) { if (record.get(TYPE) != null) { recType = EVENT_RECORD_NAMES.get(record.get(TYPE).toString()); } else { throw new ProcessingException("expected one of " + Arrays.asList(Hadoop2RecordType.values()) + " \n but not found, cannot process this record! " + jobKey); } if (recType == null) { throw new ProcessingException("new record type has surfaced: " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } Object eDetails = record.get(EVENT); if (eDetails != null) { JSONObject eventDetails = new JSONObject(eDetails.toString()); processRecords(recType, eventDetails); } else { throw new ProcessingException("expected event details but not found " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } } } catch (EOFException eof) { LOG.info("Done parsing file, reached eof for " + jobKey); } } catch (IOException ioe) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record!" + jobKey + " error: ", ioe); } catch (JSONException jse) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", jse); } catch (IllegalArgumentException iae) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", iae); } Put jobStatusPut = getJobStatusPut(); this.jobPuts.add(jobStatusPut); Put versionPut = getHadoopVersionPut(JobHistoryFileParserFactory.getHistoryFileVersion2(), this.jobKeyBytes); this.jobPuts.add(versionPut); LOG.info("For " + this.jobKey + " #jobPuts " + jobPuts.size() + " #taskPuts: " + taskPuts.size()); }
JobHistoryFileParserHadoop2 extends JobHistoryFileParserBase { @Override public void parse(byte[] historyFileContents, JobKey jobKey) throws ProcessingException { this.jobKey = jobKey; this.jobKeyBytes = jobKeyConv.toBytes(jobKey); this.jobDetails = new JobDetails(jobKey); initializeJobDetails(); setJobId(jobKey.getJobId().getJobIdString()); try { FSDataInputStream in = new FSDataInputStream(new ByteArrayWrapper(historyFileContents)); String versionIgnore = in.readLine(); this.schema = schema.parse(in.readLine()); understandSchema(schema.toString()); this.reader = new GenericDatumReader<GenericRecord>(schema); this.decoder = DecoderFactory.get().jsonDecoder(schema, in); GenericRecord record = null; Hadoop2RecordType recType = null; try { while ((record = reader.read(null, decoder)) != null) { if (record.get(TYPE) != null) { recType = EVENT_RECORD_NAMES.get(record.get(TYPE).toString()); } else { throw new ProcessingException("expected one of " + Arrays.asList(Hadoop2RecordType.values()) + " \n but not found, cannot process this record! " + jobKey); } if (recType == null) { throw new ProcessingException("new record type has surfaced: " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } Object eDetails = record.get(EVENT); if (eDetails != null) { JSONObject eventDetails = new JSONObject(eDetails.toString()); processRecords(recType, eventDetails); } else { throw new ProcessingException("expected event details but not found " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } } } catch (EOFException eof) { LOG.info("Done parsing file, reached eof for " + jobKey); } } catch (IOException ioe) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record!" + jobKey + " error: ", ioe); } catch (JSONException jse) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", jse); } catch (IllegalArgumentException iae) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", iae); } Put jobStatusPut = getJobStatusPut(); this.jobPuts.add(jobStatusPut); Put versionPut = getHadoopVersionPut(JobHistoryFileParserFactory.getHistoryFileVersion2(), this.jobKeyBytes); this.jobPuts.add(versionPut); LOG.info("For " + this.jobKey + " #jobPuts " + jobPuts.size() + " #taskPuts: " + taskPuts.size()); } }
JobHistoryFileParserHadoop2 extends JobHistoryFileParserBase { @Override public void parse(byte[] historyFileContents, JobKey jobKey) throws ProcessingException { this.jobKey = jobKey; this.jobKeyBytes = jobKeyConv.toBytes(jobKey); this.jobDetails = new JobDetails(jobKey); initializeJobDetails(); setJobId(jobKey.getJobId().getJobIdString()); try { FSDataInputStream in = new FSDataInputStream(new ByteArrayWrapper(historyFileContents)); String versionIgnore = in.readLine(); this.schema = schema.parse(in.readLine()); understandSchema(schema.toString()); this.reader = new GenericDatumReader<GenericRecord>(schema); this.decoder = DecoderFactory.get().jsonDecoder(schema, in); GenericRecord record = null; Hadoop2RecordType recType = null; try { while ((record = reader.read(null, decoder)) != null) { if (record.get(TYPE) != null) { recType = EVENT_RECORD_NAMES.get(record.get(TYPE).toString()); } else { throw new ProcessingException("expected one of " + Arrays.asList(Hadoop2RecordType.values()) + " \n but not found, cannot process this record! " + jobKey); } if (recType == null) { throw new ProcessingException("new record type has surfaced: " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } Object eDetails = record.get(EVENT); if (eDetails != null) { JSONObject eventDetails = new JSONObject(eDetails.toString()); processRecords(recType, eventDetails); } else { throw new ProcessingException("expected event details but not found " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } } } catch (EOFException eof) { LOG.info("Done parsing file, reached eof for " + jobKey); } } catch (IOException ioe) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record!" + jobKey + " error: ", ioe); } catch (JSONException jse) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", jse); } catch (IllegalArgumentException iae) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", iae); } Put jobStatusPut = getJobStatusPut(); this.jobPuts.add(jobStatusPut); Put versionPut = getHadoopVersionPut(JobHistoryFileParserFactory.getHistoryFileVersion2(), this.jobKeyBytes); this.jobPuts.add(versionPut); LOG.info("For " + this.jobKey + " #jobPuts " + jobPuts.size() + " #taskPuts: " + taskPuts.size()); } JobHistoryFileParserHadoop2(Configuration conf); }
JobHistoryFileParserHadoop2 extends JobHistoryFileParserBase { @Override public void parse(byte[] historyFileContents, JobKey jobKey) throws ProcessingException { this.jobKey = jobKey; this.jobKeyBytes = jobKeyConv.toBytes(jobKey); this.jobDetails = new JobDetails(jobKey); initializeJobDetails(); setJobId(jobKey.getJobId().getJobIdString()); try { FSDataInputStream in = new FSDataInputStream(new ByteArrayWrapper(historyFileContents)); String versionIgnore = in.readLine(); this.schema = schema.parse(in.readLine()); understandSchema(schema.toString()); this.reader = new GenericDatumReader<GenericRecord>(schema); this.decoder = DecoderFactory.get().jsonDecoder(schema, in); GenericRecord record = null; Hadoop2RecordType recType = null; try { while ((record = reader.read(null, decoder)) != null) { if (record.get(TYPE) != null) { recType = EVENT_RECORD_NAMES.get(record.get(TYPE).toString()); } else { throw new ProcessingException("expected one of " + Arrays.asList(Hadoop2RecordType.values()) + " \n but not found, cannot process this record! " + jobKey); } if (recType == null) { throw new ProcessingException("new record type has surfaced: " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } Object eDetails = record.get(EVENT); if (eDetails != null) { JSONObject eventDetails = new JSONObject(eDetails.toString()); processRecords(recType, eventDetails); } else { throw new ProcessingException("expected event details but not found " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } } } catch (EOFException eof) { LOG.info("Done parsing file, reached eof for " + jobKey); } } catch (IOException ioe) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record!" + jobKey + " error: ", ioe); } catch (JSONException jse) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", jse); } catch (IllegalArgumentException iae) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", iae); } Put jobStatusPut = getJobStatusPut(); this.jobPuts.add(jobStatusPut); Put versionPut = getHadoopVersionPut(JobHistoryFileParserFactory.getHistoryFileVersion2(), this.jobKeyBytes); this.jobPuts.add(versionPut); LOG.info("For " + this.jobKey + " #jobPuts " + jobPuts.size() + " #taskPuts: " + taskPuts.size()); } JobHistoryFileParserHadoop2(Configuration conf); @Override void parse(byte[] historyFileContents, JobKey jobKey); byte[] getTaskKey(String prefix, String jobNumber, String fullId); byte[] getAMKey(String prefix, String fullId); @Override List<Put> getJobPuts(); @Override List<Put> getTaskPuts(); void printAllPuts(List<Put> p); @Override Long getMegaByteMillis(); @Override JobDetails getJobDetails(); long getMapMbMillis(); void setMapMbMillis(long mapMbMillis); long getReduceMbMillis(); void setReduceMbMillis(long reduceMbMillis); }
JobHistoryFileParserHadoop2 extends JobHistoryFileParserBase { @Override public void parse(byte[] historyFileContents, JobKey jobKey) throws ProcessingException { this.jobKey = jobKey; this.jobKeyBytes = jobKeyConv.toBytes(jobKey); this.jobDetails = new JobDetails(jobKey); initializeJobDetails(); setJobId(jobKey.getJobId().getJobIdString()); try { FSDataInputStream in = new FSDataInputStream(new ByteArrayWrapper(historyFileContents)); String versionIgnore = in.readLine(); this.schema = schema.parse(in.readLine()); understandSchema(schema.toString()); this.reader = new GenericDatumReader<GenericRecord>(schema); this.decoder = DecoderFactory.get().jsonDecoder(schema, in); GenericRecord record = null; Hadoop2RecordType recType = null; try { while ((record = reader.read(null, decoder)) != null) { if (record.get(TYPE) != null) { recType = EVENT_RECORD_NAMES.get(record.get(TYPE).toString()); } else { throw new ProcessingException("expected one of " + Arrays.asList(Hadoop2RecordType.values()) + " \n but not found, cannot process this record! " + jobKey); } if (recType == null) { throw new ProcessingException("new record type has surfaced: " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } Object eDetails = record.get(EVENT); if (eDetails != null) { JSONObject eventDetails = new JSONObject(eDetails.toString()); processRecords(recType, eventDetails); } else { throw new ProcessingException("expected event details but not found " + record.get(TYPE).toString() + " cannot process this record! " + jobKey); } } } catch (EOFException eof) { LOG.info("Done parsing file, reached eof for " + jobKey); } } catch (IOException ioe) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record!" + jobKey + " error: ", ioe); } catch (JSONException jse) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", jse); } catch (IllegalArgumentException iae) { throw new ProcessingException(" Unable to parse history file in function parse, " + "cannot process this record! " + jobKey + " error: ", iae); } Put jobStatusPut = getJobStatusPut(); this.jobPuts.add(jobStatusPut); Put versionPut = getHadoopVersionPut(JobHistoryFileParserFactory.getHistoryFileVersion2(), this.jobKeyBytes); this.jobPuts.add(versionPut); LOG.info("For " + this.jobKey + " #jobPuts " + jobPuts.size() + " #taskPuts: " + taskPuts.size()); } JobHistoryFileParserHadoop2(Configuration conf); @Override void parse(byte[] historyFileContents, JobKey jobKey); byte[] getTaskKey(String prefix, String jobNumber, String fullId); byte[] getAMKey(String prefix, String fullId); @Override List<Put> getJobPuts(); @Override List<Put> getTaskPuts(); void printAllPuts(List<Put> p); @Override Long getMegaByteMillis(); @Override JobDetails getJobDetails(); long getMapMbMillis(); void setMapMbMillis(long mapMbMillis); long getReduceMbMillis(); void setReduceMbMillis(long reduceMbMillis); static final String JOB_STATUS_SUCCEEDED; }