MAPREDUCE-6376. Add avro binary support for jhist files. Contributed by Ray Chiang
This commit is contained in:
parent
80a68d6056
commit
2ac87df578
@ -372,6 +372,9 @@ Release 2.8.0 - UNRELEASED
|
||||
|
||||
OPTIMIZATIONS
|
||||
|
||||
MAPREDUCE-6376. Add avro binary support for jhist files (Ray Chiang via
|
||||
jlowe)
|
||||
|
||||
BUG FIXES
|
||||
|
||||
MAPREDUCE-6314. TestPipeApplication fails on trunk.
|
||||
|
@ -105,7 +105,8 @@ public class JobHistoryEventHandler extends AbstractService
|
||||
|
||||
private int numUnflushedCompletionEvents = 0;
|
||||
private boolean isTimerActive;
|
||||
|
||||
private EventWriter.WriteMode jhistMode =
|
||||
EventWriter.WriteMode.JSON;
|
||||
|
||||
protected BlockingQueue<JobHistoryEvent> eventQueue =
|
||||
new LinkedBlockingQueue<JobHistoryEvent>();
|
||||
@ -260,6 +261,20 @@ protected void serviceInit(Configuration conf) throws Exception {
|
||||
LOG.info("Emitting job history data to the timeline server is not enabled");
|
||||
}
|
||||
|
||||
// Flag for setting
|
||||
String jhistFormat = conf.get(JHAdminConfig.MR_HS_JHIST_FORMAT,
|
||||
JHAdminConfig.DEFAULT_MR_HS_JHIST_FORMAT);
|
||||
if (jhistFormat.equals("json")) {
|
||||
jhistMode = EventWriter.WriteMode.JSON;
|
||||
} else if (jhistFormat.equals("binary")) {
|
||||
jhistMode = EventWriter.WriteMode.BINARY;
|
||||
} else {
|
||||
LOG.warn("Unrecognized value '" + jhistFormat + "' for property " +
|
||||
JHAdminConfig.MR_HS_JHIST_FORMAT + ". Valid values are " +
|
||||
"'json' or 'binary'. Falling back to default value '" +
|
||||
JHAdminConfig.DEFAULT_MR_HS_JHIST_FORMAT + "'.");
|
||||
}
|
||||
|
||||
super.serviceInit(conf);
|
||||
}
|
||||
|
||||
@ -418,7 +433,7 @@ protected void serviceStop() throws Exception {
|
||||
protected EventWriter createEventWriter(Path historyFilePath)
|
||||
throws IOException {
|
||||
FSDataOutputStream out = stagingDirFS.create(historyFilePath, true);
|
||||
return new EventWriter(out);
|
||||
return new EventWriter(out, this.jhistMode);
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -190,7 +190,8 @@ private byte[] getEvents() throws Exception {
|
||||
ByteArrayOutputStream output = new ByteArrayOutputStream();
|
||||
FSDataOutputStream fsOutput = new FSDataOutputStream(output,
|
||||
new FileSystem.Statistics("scheme"));
|
||||
EventWriter writer = new EventWriter(fsOutput);
|
||||
EventWriter writer = new EventWriter(fsOutput,
|
||||
EventWriter.WriteMode.JSON);
|
||||
writer.write(getJobPriorityChangedEvent());
|
||||
writer.write(getJobStatusChangedEvent());
|
||||
writer.write(getTaskUpdatedEvent());
|
||||
|
@ -221,4 +221,11 @@ public class JHAdminConfig {
|
||||
+ "jobname.limit";
|
||||
public static final int DEFAULT_MR_HS_JOBNAME_LIMIT = 50;
|
||||
|
||||
/**
|
||||
* Settings for .jhist file format.
|
||||
*/
|
||||
public static final String MR_HS_JHIST_FORMAT =
|
||||
MR_HISTORY_PREFIX + "jhist.format";
|
||||
public static final String DEFAULT_MR_HS_JHIST_FORMAT =
|
||||
"json";
|
||||
}
|
||||
|
@ -67,15 +67,17 @@ public EventReader(DataInputStream in) throws IOException {
|
||||
this.in = in;
|
||||
this.version = in.readLine();
|
||||
|
||||
if (!EventWriter.VERSION.equals(version)) {
|
||||
throw new IOException("Incompatible event log version: "+version);
|
||||
}
|
||||
|
||||
Schema myschema = new SpecificData(Event.class.getClassLoader()).getSchema(Event.class);
|
||||
Schema.Parser parser = new Schema.Parser();
|
||||
this.schema = parser.parse(in.readLine());
|
||||
this.reader = new SpecificDatumReader(schema, myschema);
|
||||
if (EventWriter.VERSION.equals(version)) {
|
||||
this.decoder = DecoderFactory.get().jsonDecoder(schema, in);
|
||||
} else if (EventWriter.VERSION_BINARY.equals(version)) {
|
||||
this.decoder = DecoderFactory.get().binaryDecoder(in, null);
|
||||
} else {
|
||||
throw new IOException("Incompatible event log version: " + version);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -43,21 +43,38 @@
|
||||
*/
|
||||
class EventWriter {
|
||||
static final String VERSION = "Avro-Json";
|
||||
static final String VERSION_BINARY = "Avro-Binary";
|
||||
|
||||
private FSDataOutputStream out;
|
||||
private DatumWriter<Event> writer =
|
||||
new SpecificDatumWriter<Event>(Event.class);
|
||||
private Encoder encoder;
|
||||
private static final Log LOG = LogFactory.getLog(EventWriter.class);
|
||||
public enum WriteMode { JSON, BINARY }
|
||||
private final WriteMode writeMode;
|
||||
private final boolean jsonOutput; // Cache value while we have 2 modes
|
||||
|
||||
EventWriter(FSDataOutputStream out) throws IOException {
|
||||
EventWriter(FSDataOutputStream out, WriteMode mode) throws IOException {
|
||||
this.out = out;
|
||||
this.writeMode = mode;
|
||||
if (this.writeMode==WriteMode.JSON) {
|
||||
this.jsonOutput = true;
|
||||
out.writeBytes(VERSION);
|
||||
} else if (this.writeMode==WriteMode.BINARY) {
|
||||
this.jsonOutput = false;
|
||||
out.writeBytes(VERSION_BINARY);
|
||||
} else {
|
||||
throw new IOException("Unknown mode: " + mode);
|
||||
}
|
||||
out.writeBytes("\n");
|
||||
out.writeBytes(Event.SCHEMA$.toString());
|
||||
out.writeBytes("\n");
|
||||
if (!this.jsonOutput) {
|
||||
this.encoder = EncoderFactory.get().binaryEncoder(out, null);
|
||||
} else {
|
||||
this.encoder = EncoderFactory.get().jsonEncoder(Event.SCHEMA$, out);
|
||||
}
|
||||
}
|
||||
|
||||
synchronized void write(HistoryEvent event) throws IOException {
|
||||
Event wrapper = new Event();
|
||||
@ -65,8 +82,10 @@ synchronized void write(HistoryEvent event) throws IOException {
|
||||
wrapper.setEvent(event.getDatum());
|
||||
writer.write(wrapper, encoder);
|
||||
encoder.flush();
|
||||
if (this.jsonOutput) {
|
||||
out.writeBytes("\n");
|
||||
}
|
||||
}
|
||||
|
||||
void flush() throws IOException {
|
||||
encoder.flush();
|
||||
|
@ -1713,6 +1713,15 @@
|
||||
</description>
|
||||
</property>
|
||||
|
||||
<property>
|
||||
<description>
|
||||
File format the AM will use when generating the .jhist file. Valid
|
||||
values are "json" for text output and "binary" for faster parsing.
|
||||
</description>
|
||||
<name>mapreduce.jobhistory.jhist.format</name>
|
||||
<value>json</value>
|
||||
</property>
|
||||
|
||||
<property>
|
||||
<name>mapreduce.job.heap.memory-mb.ratio</name>
|
||||
<value>0.8</value>
|
||||
|
Loading…
Reference in New Issue
Block a user