@Test public void testWrite() throws Exception { Schema schema = new Schema.Parser().parse(this.getClass().getClassLoader().getResourceAsStream("orc_writer_test/schema.avsc")); List<GenericRecord> recordList = deserializeAvroRecords(this.getClass(), schema, "orc_writer_test/data.json"); FsDataWriterBuilder<Schema, GenericRecord> mockBuilder = (FsDataWriterBuilder<Schema, GenericRecord>) Mockito.mock(FsDataWriterBuilder.class); when(mockBuilder.getSchema()).thenReturn(schema); State dummyState = new WorkUnit(); String stagingDir = Files.createTempDir().getAbsolutePath(); String outputDir = Files.createTempDir().getAbsolutePath(); dummyState.setProp(ConfigurationKeys.WRITER_STAGING_DIR, stagingDir); dummyState.setProp(ConfigurationKeys.WRITER_FILE_PATH, "simple"); dummyState.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, outputDir); when(mockBuilder.getFileName(dummyState)).thenReturn("file"); Path outputFilePath = new Path(outputDir, "simple/file"); Closer closer = Closer.create(); GobblinOrcWriter orcWriter = closer.register(new GobblinOrcWriter(mockBuilder, dummyState)); GobblinOrcWriter orcFailWriter = new GobblinOrcWriter(mockBuilder, dummyState); for (GenericRecord record : recordList) { orcWriter.write(record); orcFailWriter.write(record); } Assert.assertEquals(orcWriter.recordsWritten(), 0); Assert.assertEquals(orcFailWriter.recordsWritten(), 0); try { orcFailWriter.close(); } catch (CloseBeforeFlushException e) { Assert.assertEquals(e.datasetName, schema.getName()); } orcWriter.commit(); Assert.assertEquals(orcWriter.recordsWritten(), 2); FileSystem fs = FileSystem.getLocal(new Configuration()); Assert.assertTrue(fs.exists(outputFilePath)); List<Writable> orcRecords = deserializeOrcRecords(outputFilePath, fs); Assert.assertEquals(orcRecords.size(), 2); try { closer.close(); } catch (NullPointerException npe) { Assert.fail(); } }
@Override public void write(GenericRecord record) throws IOException { valueWriter.write(record, rowBatch); if (rowBatch.size == this.batchSize) { orcFileWriter.addRowBatch(rowBatch); rowBatch.reset(); if (deepCleanBatch) { log.info("A reset of rowBatch is triggered - releasing holding memory for large object"); deepCleanRowBatch(rowBatch); } } }
GobblinOrcWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { valueWriter.write(record, rowBatch); if (rowBatch.size == this.batchSize) { orcFileWriter.addRowBatch(rowBatch); rowBatch.reset(); if (deepCleanBatch) { log.info("A reset of rowBatch is triggered - releasing holding memory for large object"); deepCleanRowBatch(rowBatch); } } } }
GobblinOrcWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { valueWriter.write(record, rowBatch); if (rowBatch.size == this.batchSize) { orcFileWriter.addRowBatch(rowBatch); rowBatch.reset(); if (deepCleanBatch) { log.info("A reset of rowBatch is triggered - releasing holding memory for large object"); deepCleanRowBatch(rowBatch); } } } GobblinOrcWriter(FsDataWriterBuilder<Schema, GenericRecord> builder, State properties); }
GobblinOrcWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { valueWriter.write(record, rowBatch); if (rowBatch.size == this.batchSize) { orcFileWriter.addRowBatch(rowBatch); rowBatch.reset(); if (deepCleanBatch) { log.info("A reset of rowBatch is triggered - releasing holding memory for large object"); deepCleanRowBatch(rowBatch); } } } GobblinOrcWriter(FsDataWriterBuilder<Schema, GenericRecord> builder, State properties); @Override long recordsWritten(); @Override State getFinalState(); @Override void flush(); @Override void close(); @Override void commit(); @Override void write(GenericRecord record); @Override boolean isSpeculativeAttemptSafe(); static Properties getPropsWithOrcSchema(Schema avroSchema); }
GobblinOrcWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { valueWriter.write(record, rowBatch); if (rowBatch.size == this.batchSize) { orcFileWriter.addRowBatch(rowBatch); rowBatch.reset(); if (deepCleanBatch) { log.info("A reset of rowBatch is triggered - releasing holding memory for large object"); deepCleanRowBatch(rowBatch); } } } GobblinOrcWriter(FsDataWriterBuilder<Schema, GenericRecord> builder, State properties); @Override long recordsWritten(); @Override State getFinalState(); @Override void flush(); @Override void close(); @Override void commit(); @Override void write(GenericRecord record); @Override boolean isSpeculativeAttemptSafe(); static Properties getPropsWithOrcSchema(Schema avroSchema);  }