@Test public void testWrite() throws IOException { for (String record : TestConstants.JSON_RECORDS) { this.writer.write(convertRecord(record)); } Assert.assertEquals(this.writer.recordsWritten(), 3); this.writer.close(); this.writer.commit(); File outputFile = new File(TestConstants.TEST_OUTPUT_DIR + Path.SEPARATOR + this.filePath, TestConstants.TEST_FILE_NAME); DataFileReader<GenericRecord> reader = new DataFileReader<>(outputFile, new GenericDatumReader<GenericRecord>()); Schema fileSchema = reader.getSchema(); Assert.assertEquals(fileSchema.getProp(TEST_PROPERTY_KEY), TEST_PROPERTY_VALUE); GenericRecord user1 = reader.next(); Assert.assertEquals(user1.get("name").toString(), "Alyssa"); Assert.assertEquals(user1.get("favorite_number"), 256); Assert.assertEquals(user1.get("favorite_color").toString(), "yellow"); GenericRecord user2 = reader.next(); Assert.assertEquals(user2.get("name").toString(), "Ben"); Assert.assertEquals(user2.get("favorite_number"), 7); Assert.assertEquals(user2.get("favorite_color").toString(), "red"); GenericRecord user3 = reader.next(); Assert.assertEquals(user3.get("name").toString(), "Charlie"); Assert.assertEquals(user3.get("favorite_number"), 68); Assert.assertEquals(user3.get("favorite_color").toString(), "blue"); reader.close(); FsWriterMetrics metrics = FsWriterMetrics.fromJson(properties.getProp(FsDataWriter.FS_WRITER_METRICS_KEY)); Assert.assertEquals(metrics.fileInfos.size(),1); FsWriterMetrics.FileInfo fileInfo = metrics.fileInfos.iterator().next(); Assert.assertEquals(fileInfo.fileName, TestConstants.TEST_FILE_NAME); Assert.assertEquals(fileInfo.numRecords, 3); Assert.assertNull(metrics.partitionInfo.partitionKey); Assert.assertEquals(metrics.partitionInfo.branchId, 0); }
@Override public void write(GenericRecord record) throws IOException { if (skipNullRecord && record == null) { return; } Preconditions.checkNotNull(record); this.writer.append(record); this.count.incrementAndGet(); }
AvroHdfsDataWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { if (skipNullRecord && record == null) { return; } Preconditions.checkNotNull(record); this.writer.append(record); this.count.incrementAndGet(); } }
AvroHdfsDataWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { if (skipNullRecord && record == null) { return; } Preconditions.checkNotNull(record); this.writer.append(record); this.count.incrementAndGet(); } AvroHdfsDataWriter(FsDataWriterBuilder<Schema, GenericRecord> builder, State state); }
AvroHdfsDataWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { if (skipNullRecord && record == null) { return; } Preconditions.checkNotNull(record); this.writer.append(record); this.count.incrementAndGet(); } AvroHdfsDataWriter(FsDataWriterBuilder<Schema, GenericRecord> builder, State state); FileSystem getFileSystem(); @Override void write(GenericRecord record); @Override long recordsWritten(); @Override boolean isSpeculativeAttemptSafe(); @Override void flush(); }
AvroHdfsDataWriter extends FsDataWriter<GenericRecord> { @Override public void write(GenericRecord record) throws IOException { if (skipNullRecord && record == null) { return; } Preconditions.checkNotNull(record); this.writer.append(record); this.count.incrementAndGet(); } AvroHdfsDataWriter(FsDataWriterBuilder<Schema, GenericRecord> builder, State state); FileSystem getFileSystem(); @Override void write(GenericRecord record); @Override long recordsWritten(); @Override boolean isSpeculativeAttemptSafe(); @Override void flush();  }