From 52e1b917d7045e5e12853aaea8eb66c6a4df8b3f Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Fri, 2 Oct 2026 23:02:36 -0700 Subject: [PATCH] more robust metrics --- .../io/iceberg/cdc/sink/CommitterMetrics.java | 73 +++++++------------ .../sdk/io/iceberg/cdc/sink/package-info.java | 10 +-- .../io/iceberg/cdc/sink/CommitDeltasTest.java | 34 +++++++-- .../cdc/sink/CommitterMetricsTest.java | 27 +++---- 4 files changed, 70 insertions(+), 74 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetrics.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetrics.java index 63d2a13c701b..c7330386f5d3 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetrics.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetrics.java @@ -33,12 +33,14 @@ final class CommitterMetrics implements Serializable { final Counter snapshotsCreated = Metrics.counter(CommitDeltas.class, "snapshotsCreated"); - final Counter committedDataFiles = Metrics.counter(CommitDeltas.class, "committedDataFiles"); - final Counter committedDeleteFiles = Metrics.counter(CommitDeltas.class, "committedDeleteFiles"); - final Counter committedRecords = Metrics.counter(CommitDeltas.class, "committedRecords"); - final Counter committedEqualityDeleteRecords = - Metrics.counter(CommitDeltas.class, "committedEqualityDeleteRecords"); - final Counter committedBytes = Metrics.counter(CommitDeltas.class, "committedBytes"); + final Distribution committedDataFileRecordCount = + Metrics.distribution(CommitDeltas.class, "committedDataFileRecordCount"); + final Distribution committedDataFileByteSize = + Metrics.distribution(CommitDeltas.class, "committedDataFileByteSize"); + final Distribution committedEqualityDeleteRecordCount = + Metrics.distribution(CommitDeltas.class, "committedEqualityDeleteRecordCount"); + final Distribution committedEqualityDeleteByteSize = + Metrics.distribution(CommitDeltas.class, "committedEqualityDeleteByteSize"); final Distribution commitDurationMs = Metrics.distribution(CommitDeltas.class, "commitDurationMs"); final Counter commitFailures = Metrics.counter(CommitDeltas.class, "commitFailures"); @@ -57,79 +59,58 @@ final class CommitterMetrics implements Serializable { /** Records a successful commit's volume metrics. */ void recordCommit(CommitSummary summary) { - committedDataFiles.inc(summary.dataFileCount); - committedDeleteFiles.inc(summary.deleteFileCount); - committedRecords.inc(summary.dataRecords); - committedEqualityDeleteRecords.inc(summary.equalityDeleteRecords); - committedBytes.inc(summary.bytes); + for (DataFile f : summary.dataFiles) { + committedDataFileRecordCount.update(f.recordCount()); + committedDataFileByteSize.update(f.fileSizeInBytes()); + } + for (DeleteFile f : summary.deleteFiles) { + if (f.content() == FileContent.EQUALITY_DELETES) { + committedEqualityDeleteRecordCount.update(f.recordCount()); + committedEqualityDeleteByteSize.update(f.fileSizeInBytes()); + } + } snapshotsCreated.inc(); } - /** One commit's volume and partition-spec summary, computed from its files. */ + /** One commit's files and partition-spec summary. */ static final class CommitSummary { - final long dataFileCount; - final long deleteFileCount; - final long dataRecords; - final long equalityDeleteRecords; - final long bytes; + final List dataFiles; + final List deleteFiles; final @Nullable Integer firstSpecId; final Set specIds; final boolean hasEqualityDeletes; private CommitSummary( - long dataFileCount, - long deleteFileCount, - long dataRecords, - long equalityDeleteRecords, - long bytes, + List dataFiles, + List deleteFiles, @Nullable Integer firstSpecId, Set specIds, boolean hasEqualityDeletes) { - this.dataFileCount = dataFileCount; - this.deleteFileCount = deleteFileCount; - this.dataRecords = dataRecords; - this.equalityDeleteRecords = equalityDeleteRecords; - this.bytes = bytes; + this.dataFiles = dataFiles; + this.deleteFiles = deleteFiles; this.firstSpecId = firstSpecId; this.specIds = specIds; this.hasEqualityDeletes = hasEqualityDeletes; } static CommitSummary of(List dataFiles, List deleteFiles) { - long dataRecords = 0; - long bytes = 0; @Nullable Integer firstSpecId = null; Set specIds = new HashSet<>(); for (DataFile f : dataFiles) { - dataRecords += f.recordCount(); - bytes += f.fileSizeInBytes(); if (firstSpecId == null) { firstSpecId = f.specId(); } specIds.add(f.specId()); } - long equalityDeleteRecords = 0; boolean hasEqualityDeletes = false; for (DeleteFile f : deleteFiles) { - bytes += f.fileSizeInBytes(); if (firstSpecId == null) { firstSpecId = f.specId(); } specIds.add(f.specId()); - if (f.content() == FileContent.EQUALITY_DELETES) { - equalityDeleteRecords += f.recordCount(); - hasEqualityDeletes = true; - } + hasEqualityDeletes |= f.content() == FileContent.EQUALITY_DELETES; } - return new CommitSummary( - dataFiles.size(), - deleteFiles.size(), - dataRecords, - equalityDeleteRecords, - bytes, - firstSpecId, - specIds, - hasEqualityDeletes); + return new CommitSummary(dataFiles, deleteFiles, firstSpecId, specIds, hasEqualityDeletes); } } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java index c25496648f98..31786596d6c3 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java @@ -41,12 +41,12 @@ *
  • snapshotsCreated: CDC snapshots committed. In streaming this should track one per * commit window per destination. A flat line while data is flowing means commits have * stalled. - *
  • committedDataFiles / committedDeleteFiles: files published. Divide by {@code - * snapshotsCreated} for files per commit. A large ratio points to the small-file problem + *
  • committedDataFileRecordCount / committedEqualityDeleteRecordCount + * (distributions, per file): rows in each committed data file and equality-delete file. Their + * counts are the files committed; divide by {@code snapshotsCreated} for files per commit. + *
  • committedDataFileByteSize / committedEqualityDeleteByteSize (distributions, + * per file): size of each committed file. A low mean points to the small-file problem * described below. - *
  • committedRecords: rows committed. - *
  • committedEqualityDeleteRecords: equality-delete rows committed. - *
  • committedBytes: total bytes of committed files. *
  • commitDurationMs (distribution): wall-clock time spent waiting on the Iceberg commit * operation. Its tail is your catalog's health: a growing maximum usually means catalog * contention, not a Beam problem. diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java index 95dd70de3282..38b38ca1b4d9 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java @@ -56,6 +56,7 @@ import org.apache.beam.sdk.io.iceberg.SerializableDataFile; import org.apache.beam.sdk.io.iceberg.SerializableDeleteFile; import org.apache.beam.sdk.io.iceberg.SnapshotInfo; +import org.apache.beam.sdk.metrics.DistributionResult; import org.apache.beam.sdk.metrics.MetricNameFilter; import org.apache.beam.sdk.metrics.MetricResult; import org.apache.beam.sdk.metrics.Metrics; @@ -302,6 +303,19 @@ private static long counter(PipelineResult result, String name) { return total; } + /** The named {@link CommitDeltas} distribution committed by the pipeline. */ + private static DistributionResult distribution(PipelineResult result, String name) { + return Iterables.getOnlyElement( + result + .metrics() + .queryMetrics( + MetricsFilter.builder() + .addNameFilter(MetricNameFilter.named(CommitDeltas.class, name)) + .build()) + .getDistributions()) + .getCommitted(); + } + /** * Seeds a real prior CDC snapshot for {@code sinkId}, token keys written as the committer would. * The keys are literal strings on purpose: they double as a pin of the on-disk contract. @@ -1243,7 +1257,7 @@ public void describeSkippedFilesListsUpToFivePathsAndFallsThroughToDeleteFiles() // Commit volume metrics. // --------------------------------------------------------------------------------------------- - /** A commit records its volume metrics: file/record counts, bytes, snapshots created. */ + /** A commit records its volume metrics: records and bytes per committed file. */ @Test public void recordsCommitVolumeMetrics() { TestTable tt = v2Table(); @@ -1283,12 +1297,18 @@ public void recordsCommitVolumeMetrics() { result.waitUntilFinish(); assertThat(counter(result, "snapshotsCreated"), equalTo(1L)); - assertThat(counter(result, "committedDataFiles"), equalTo(2L)); - assertThat(counter(result, "committedDeleteFiles"), equalTo(2L)); - // 2 data files carry 1 record each (id=1, id=2). - assertThat(counter(result, "committedRecords"), equalTo(2L)); - assertThat(counter(result, "committedEqualityDeleteRecords"), equalTo(2L)); - assertThat(counter(result, "committedBytes"), greaterThan(0L)); + // One commit with 2 data files and 2 equality-delete files, each carrying 1 record (rows + // id=1, id=2; deletes for id=1, id=3). + assertThat( + distribution(result, "committedDataFileRecordCount"), + equalTo(DistributionResult.create(2, 2, 1, 1))); + assertThat( + distribution(result, "committedEqualityDeleteRecordCount"), + equalTo(DistributionResult.create(2, 2, 1, 1))); + for (String bytes : List.of("committedDataFileByteSize", "committedEqualityDeleteByteSize")) { + assertThat(distribution(result, bytes).getCount(), equalTo(2L)); + assertThat(distribution(result, bytes).getMin(), greaterThan(0L)); + } } // --------------------------------------------------------------------------------------------- diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetricsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetricsTest.java index 9f8f73f73195..ce89570188fb 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetricsTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitterMetricsTest.java @@ -69,20 +69,17 @@ private static DeleteFile deleteFile( .build(); } - // Counts, records, and bytes cover both file kinds; only equality deletes count as delete - // records. + // The summary keeps both file lists for the per-file metrics. @Test - public void summarizesCountsRecordsAndBytes() { - CommitSummary summary = - CommitSummary.of( - ImmutableList.of(dataFile(SPEC_0, 10, 100), dataFile(SPEC_0, 20, 200)), - ImmutableList.of(deleteFile(SPEC_0, true, 4, 40), deleteFile(SPEC_0, false, 7, 70))); + public void keepsFilesAndFlagsEqualityDeletes() { + ImmutableList dataFiles = + ImmutableList.of(dataFile(SPEC_0, 10, 100), dataFile(SPEC_0, 20, 200)); + ImmutableList deleteFiles = + ImmutableList.of(deleteFile(SPEC_0, true, 4, 40), deleteFile(SPEC_0, false, 7, 70)); + CommitSummary summary = CommitSummary.of(dataFiles, deleteFiles); - assertThat(summary.dataFileCount, equalTo(2L)); - assertThat(summary.deleteFileCount, equalTo(2L)); - assertThat(summary.dataRecords, equalTo(30L)); - assertThat(summary.equalityDeleteRecords, equalTo(4L)); - assertThat(summary.bytes, equalTo(410L)); + assertThat(summary.dataFiles, equalTo(dataFiles)); + assertThat(summary.deleteFiles, equalTo(deleteFiles)); assertThat(summary.hasEqualityDeletes, equalTo(true)); } @@ -92,9 +89,7 @@ public void positionDeletesAreNotEqualityDeletes() { CommitSummary summary = CommitSummary.of(ImmutableList.of(), ImmutableList.of(deleteFile(SPEC_0, false, 7, 70))); - assertThat(summary.equalityDeleteRecords, equalTo(0L)); assertThat(summary.hasEqualityDeletes, equalTo(false)); - assertThat(summary.bytes, equalTo(70L)); } // The first spec id comes from the data files when there are any, and every spec id is collected. @@ -116,8 +111,8 @@ public void firstSpecIdPrefersDataFilesAndSpecIdsCollectsAll() { public void emptyWindowHasNoSpec() { CommitSummary summary = CommitSummary.of(ImmutableList.of(), ImmutableList.of()); - assertThat(summary.dataFileCount, equalTo(0L)); - assertThat(summary.bytes, equalTo(0L)); + assertThat(summary.dataFiles, empty()); + assertThat(summary.deleteFiles, empty()); assertThat(summary.firstSpecId, nullValue()); assertThat(summary.specIds, empty()); }