@Test public void testDataInternalWriter() throws IOException { HoodieWriteConfig cfg = getConfigBuilder(basePath).build(); HoodieTable table = HoodieSparkTable.create(cfg, context, metaClient); for (int i = 0; i < 5; i++) { String instantTime = "00" + i; HoodieBulkInsertDataInternalWriter writer = new HoodieBulkInsertDataInternalWriter(table, cfg, instantTime, RANDOM.nextInt(100000), RANDOM.nextLong(), RANDOM.nextLong(), STRUCT_TYPE); int size = 10 + RANDOM.nextInt(1000); int batches = 5; Dataset<Row> totalInputRows = null; for (int j = 0; j < batches; j++) { String partitionPath = HoodieTestDataGenerator.DEFAULT_PARTITION_PATHS[j % 3]; Dataset<Row> inputRows = getRandomRows(sqlContext, size, partitionPath, false); writeRows(inputRows, writer); if (totalInputRows == null) { totalInputRows = inputRows; } else { totalInputRows = totalInputRows.union(inputRows); } } HoodieWriterCommitMessage commitMetadata = (HoodieWriterCommitMessage) writer.commit(); List<String> fileAbsPaths = new ArrayList<>(); List<String> fileNames = new ArrayList<>(); assertWriteStatuses(commitMetadata.getWriteStatuses(), batches, size, fileAbsPaths, fileNames); Dataset<Row> result = sqlContext.read().parquet(fileAbsPaths.toArray(new String[0])); assertOutput(totalInputRows, result, instantTime, fileNames); } }
@Override public WriterCommitMessage commit() throws IOException { close(); return new HoodieWriterCommitMessage(writeStatusList); }
HoodieBulkInsertDataInternalWriter implements DataWriter<InternalRow> { @Override public WriterCommitMessage commit() throws IOException { close(); return new HoodieWriterCommitMessage(writeStatusList); } }
HoodieBulkInsertDataInternalWriter implements DataWriter<InternalRow> { @Override public WriterCommitMessage commit() throws IOException { close(); return new HoodieWriterCommitMessage(writeStatusList); } HoodieBulkInsertDataInternalWriter(HoodieTable hoodieTable, HoodieWriteConfig writeConfig,
      String instantTime, int taskPartitionId, long taskId, long taskEpochId,
      StructType structType); }
HoodieBulkInsertDataInternalWriter implements DataWriter<InternalRow> { @Override public WriterCommitMessage commit() throws IOException { close(); return new HoodieWriterCommitMessage(writeStatusList); } HoodieBulkInsertDataInternalWriter(HoodieTable hoodieTable, HoodieWriteConfig writeConfig,
      String instantTime, int taskPartitionId, long taskId, long taskEpochId,
      StructType structType); @Override void write(InternalRow record); @Override WriterCommitMessage commit(); @Override void abort(); void close(); }
HoodieBulkInsertDataInternalWriter implements DataWriter<InternalRow> { @Override public WriterCommitMessage commit() throws IOException { close(); return new HoodieWriterCommitMessage(writeStatusList); } HoodieBulkInsertDataInternalWriter(HoodieTable hoodieTable, HoodieWriteConfig writeConfig,
      String instantTime, int taskPartitionId, long taskId, long taskEpochId,
      StructType structType); @Override void write(InternalRow record); @Override WriterCommitMessage commit(); @Override void abort(); void close();  }