@Test public void testWriteAmplificationStats() throws Exception { Map<String, Integer[]> data = new LinkedHashMap<>(); data.put("100", new Integer[] {15, 10}); data.put("101", new Integer[] {20, 10}); data.put("102", new Integer[] {15, 15}); for (Map.Entry<String, Integer[]> entry : data.entrySet()) { String k = entry.getKey(); Integer[] v = entry.getValue(); HoodieTestCommitMetadataGenerator.createCommitFileWithMetadata(tablePath, k, jsc.hadoopConfiguration(), Option.of(v[0]), Option.of(v[1])); } CommandResult cr = getShell().executeCommand("stats wa"); assertTrue(cr.isSuccess()); List<Comparable[]> rows = new ArrayList<>(); DecimalFormat df = new DecimalFormat("#.00"); data.forEach((key, value) -> { rows.add(new Comparable[] {key, value[1] * 2, value[0] * 2, df.format((float) value[0] / value[1])}); }); int totalWrite = data.values().stream().map(integers -> integers[0] * 2).mapToInt(s -> s).sum(); int totalUpdate = data.values().stream().map(integers -> integers[1] * 2).mapToInt(s -> s).sum(); rows.add(new Comparable[]{"Total", totalUpdate, totalWrite, df.format((float) totalWrite / totalUpdate)}); TableHeader header = new TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_UPSERTED) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_WRITTEN) .addTableHeaderField(HoodieTableHeaderFields.HEADER_WRITE_AMPLIFICATION_FACTOR); String expected = HoodiePrintHelper.print(header, new HashMap<>(), "", false, -1, false, rows); expected = removeNonWordAndStripSpace(expected); String got = removeNonWordAndStripSpace(cr.getResult().toString()); assertEquals(expected, got); }
@CliCommand(value = "stats wa", help = "Write Amplification. Ratio of how many records were upserted to how many " + "records were actually written") public String writeAmplificationStats( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { long totalRecordsUpserted = 0; long totalRecordsWritten = 0; HoodieActiveTimeline activeTimeline = HoodieCLI.getTableMetaClient().getActiveTimeline(); HoodieTimeline timeline = activeTimeline.getCommitTimeline().filterCompletedInstants(); List<Comparable[]> rows = new ArrayList<>(); DecimalFormat df = new DecimalFormat("#.00"); for (HoodieInstant instantTime : timeline.getInstants().collect(Collectors.toList())) { String waf = "0"; HoodieCommitMetadata commit = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(instantTime).get(), HoodieCommitMetadata.class); if (commit.fetchTotalUpdateRecordsWritten() > 0) { waf = df.format((float) commit.fetchTotalRecordsWritten() / commit.fetchTotalUpdateRecordsWritten()); } rows.add(new Comparable[] {instantTime.getTimestamp(), commit.fetchTotalUpdateRecordsWritten(), commit.fetchTotalRecordsWritten(), waf}); totalRecordsUpserted += commit.fetchTotalUpdateRecordsWritten(); totalRecordsWritten += commit.fetchTotalRecordsWritten(); } String waf = "0"; if (totalRecordsUpserted > 0) { waf = df.format((float) totalRecordsWritten / totalRecordsUpserted); } rows.add(new Comparable[] {"Total", totalRecordsUpserted, totalRecordsWritten, waf}); TableHeader header = new TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_UPSERTED) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_WRITTEN) .addTableHeaderField(HoodieTableHeaderFields.HEADER_WRITE_AMPLIFICATION_FACTOR); return HoodiePrintHelper.print(header, new HashMap<>(), sortByField, descending, limit, headerOnly, rows); }
StatsCommand implements CommandMarker { @CliCommand(value = "stats wa", help = "Write Amplification. Ratio of how many records were upserted to how many " + "records were actually written") public String writeAmplificationStats( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { long totalRecordsUpserted = 0; long totalRecordsWritten = 0; HoodieActiveTimeline activeTimeline = HoodieCLI.getTableMetaClient().getActiveTimeline(); HoodieTimeline timeline = activeTimeline.getCommitTimeline().filterCompletedInstants(); List<Comparable[]> rows = new ArrayList<>(); DecimalFormat df = new DecimalFormat("#.00"); for (HoodieInstant instantTime : timeline.getInstants().collect(Collectors.toList())) { String waf = "0"; HoodieCommitMetadata commit = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(instantTime).get(), HoodieCommitMetadata.class); if (commit.fetchTotalUpdateRecordsWritten() > 0) { waf = df.format((float) commit.fetchTotalRecordsWritten() / commit.fetchTotalUpdateRecordsWritten()); } rows.add(new Comparable[] {instantTime.getTimestamp(), commit.fetchTotalUpdateRecordsWritten(), commit.fetchTotalRecordsWritten(), waf}); totalRecordsUpserted += commit.fetchTotalUpdateRecordsWritten(); totalRecordsWritten += commit.fetchTotalRecordsWritten(); } String waf = "0"; if (totalRecordsUpserted > 0) { waf = df.format((float) totalRecordsWritten / totalRecordsUpserted); } rows.add(new Comparable[] {"Total", totalRecordsUpserted, totalRecordsWritten, waf}); TableHeader header = new TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_UPSERTED) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_WRITTEN) .addTableHeaderField(HoodieTableHeaderFields.HEADER_WRITE_AMPLIFICATION_FACTOR); return HoodiePrintHelper.print(header, new HashMap<>(), sortByField, descending, limit, headerOnly, rows); } }
StatsCommand implements CommandMarker { @CliCommand(value = "stats wa", help = "Write Amplification. Ratio of how many records were upserted to how many " + "records were actually written") public String writeAmplificationStats( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { long totalRecordsUpserted = 0; long totalRecordsWritten = 0; HoodieActiveTimeline activeTimeline = HoodieCLI.getTableMetaClient().getActiveTimeline(); HoodieTimeline timeline = activeTimeline.getCommitTimeline().filterCompletedInstants(); List<Comparable[]> rows = new ArrayList<>(); DecimalFormat df = new DecimalFormat("#.00"); for (HoodieInstant instantTime : timeline.getInstants().collect(Collectors.toList())) { String waf = "0"; HoodieCommitMetadata commit = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(instantTime).get(), HoodieCommitMetadata.class); if (commit.fetchTotalUpdateRecordsWritten() > 0) { waf = df.format((float) commit.fetchTotalRecordsWritten() / commit.fetchTotalUpdateRecordsWritten()); } rows.add(new Comparable[] {instantTime.getTimestamp(), commit.fetchTotalUpdateRecordsWritten(), commit.fetchTotalRecordsWritten(), waf}); totalRecordsUpserted += commit.fetchTotalUpdateRecordsWritten(); totalRecordsWritten += commit.fetchTotalRecordsWritten(); } String waf = "0"; if (totalRecordsUpserted > 0) { waf = df.format((float) totalRecordsWritten / totalRecordsUpserted); } rows.add(new Comparable[] {"Total", totalRecordsUpserted, totalRecordsWritten, waf}); TableHeader header = new TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_UPSERTED) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_WRITTEN) .addTableHeaderField(HoodieTableHeaderFields.HEADER_WRITE_AMPLIFICATION_FACTOR); return HoodiePrintHelper.print(header, new HashMap<>(), sortByField, descending, limit, headerOnly, rows); }  }
StatsCommand implements CommandMarker { @CliCommand(value = "stats wa", help = "Write Amplification. Ratio of how many records were upserted to how many " + "records were actually written") public String writeAmplificationStats( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { long totalRecordsUpserted = 0; long totalRecordsWritten = 0; HoodieActiveTimeline activeTimeline = HoodieCLI.getTableMetaClient().getActiveTimeline(); HoodieTimeline timeline = activeTimeline.getCommitTimeline().filterCompletedInstants(); List<Comparable[]> rows = new ArrayList<>(); DecimalFormat df = new DecimalFormat("#.00"); for (HoodieInstant instantTime : timeline.getInstants().collect(Collectors.toList())) { String waf = "0"; HoodieCommitMetadata commit = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(instantTime).get(), HoodieCommitMetadata.class); if (commit.fetchTotalUpdateRecordsWritten() > 0) { waf = df.format((float) commit.fetchTotalRecordsWritten() / commit.fetchTotalUpdateRecordsWritten()); } rows.add(new Comparable[] {instantTime.getTimestamp(), commit.fetchTotalUpdateRecordsWritten(), commit.fetchTotalRecordsWritten(), waf}); totalRecordsUpserted += commit.fetchTotalUpdateRecordsWritten(); totalRecordsWritten += commit.fetchTotalRecordsWritten(); } String waf = "0"; if (totalRecordsUpserted > 0) { waf = df.format((float) totalRecordsWritten / totalRecordsUpserted); } rows.add(new Comparable[] {"Total", totalRecordsUpserted, totalRecordsWritten, waf}); TableHeader header = new TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_UPSERTED) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_WRITTEN) .addTableHeaderField(HoodieTableHeaderFields.HEADER_WRITE_AMPLIFICATION_FACTOR); return HoodiePrintHelper.print(header, new HashMap<>(), sortByField, descending, limit, headerOnly, rows); }  @CliCommand(value = "stats wa", help = "Write Amplification. Ratio of how many records were upserted to how many " + "records were actually written") String writeAmplificationStats(
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); Comparable[] printFileSizeHistogram(String instantTime, Snapshot s); @CliCommand(value = "stats filesizes", help = "File Sizes. Display summary stats on sizes of files") String fileSizeStats(
      @CliOption(key = {"partitionPath"}, help = "regex to select files, eg: 2016/08/02",
          unspecifiedDefaultValue = "*/*/*") final String globRegex,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); Map<String, Function<Object, String>> getFieldNameToConverterMap(); }
StatsCommand implements CommandMarker { @CliCommand(value = "stats wa", help = "Write Amplification. Ratio of how many records were upserted to how many " + "records were actually written") public String writeAmplificationStats( @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit, @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField, @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending, @CliOption(key = {"headeronly"}, help = "Print Header Only", unspecifiedDefaultValue = "false") final boolean headerOnly) throws IOException { long totalRecordsUpserted = 0; long totalRecordsWritten = 0; HoodieActiveTimeline activeTimeline = HoodieCLI.getTableMetaClient().getActiveTimeline(); HoodieTimeline timeline = activeTimeline.getCommitTimeline().filterCompletedInstants(); List<Comparable[]> rows = new ArrayList<>(); DecimalFormat df = new DecimalFormat("#.00"); for (HoodieInstant instantTime : timeline.getInstants().collect(Collectors.toList())) { String waf = "0"; HoodieCommitMetadata commit = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(instantTime).get(), HoodieCommitMetadata.class); if (commit.fetchTotalUpdateRecordsWritten() > 0) { waf = df.format((float) commit.fetchTotalRecordsWritten() / commit.fetchTotalUpdateRecordsWritten()); } rows.add(new Comparable[] {instantTime.getTimestamp(), commit.fetchTotalUpdateRecordsWritten(), commit.fetchTotalRecordsWritten(), waf}); totalRecordsUpserted += commit.fetchTotalUpdateRecordsWritten(); totalRecordsWritten += commit.fetchTotalRecordsWritten(); } String waf = "0"; if (totalRecordsUpserted > 0) { waf = df.format((float) totalRecordsWritten / totalRecordsUpserted); } rows.add(new Comparable[] {"Total", totalRecordsUpserted, totalRecordsWritten, waf}); TableHeader header = new TableHeader().addTableHeaderField(HoodieTableHeaderFields.HEADER_COMMIT_TIME) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_UPSERTED) .addTableHeaderField(HoodieTableHeaderFields.HEADER_TOTAL_WRITTEN) .addTableHeaderField(HoodieTableHeaderFields.HEADER_WRITE_AMPLIFICATION_FACTOR); return HoodiePrintHelper.print(header, new HashMap<>(), sortByField, descending, limit, headerOnly, rows); }  @CliCommand(value = "stats wa", help = "Write Amplification. Ratio of how many records were upserted to how many " + "records were actually written") String writeAmplificationStats(
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); Comparable[] printFileSizeHistogram(String instantTime, Snapshot s); @CliCommand(value = "stats filesizes", help = "File Sizes. Display summary stats on sizes of files") String fileSizeStats(
      @CliOption(key = {"partitionPath"}, help = "regex to select files, eg: 2016/08/02",
          unspecifiedDefaultValue = "*/*/*") final String globRegex,
      @CliOption(key = {"limit"}, help = "Limit commits", unspecifiedDefaultValue = "-1") final Integer limit,
      @CliOption(key = {"sortBy"}, help = "Sorting Field", unspecifiedDefaultValue = "") final String sortByField,
      @CliOption(key = {"desc"}, help = "Ordering", unspecifiedDefaultValue = "false") final boolean descending,
      @CliOption(key = {"headeronly"}, help = "Print Header Only",
          unspecifiedDefaultValue = "false") final boolean headerOnly); Map<String, Function<Object, String>> getFieldNameToConverterMap(); static final int MAX_FILES; }