This is an automated email from the ASF dual-hosted git repository.
ahmedabu98 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 33547f6ccad more robust metrics (#40394)
33547f6ccad is described below
commit 33547f6ccad486a819e895fe4a91fbbd93666856
Author: Ahmed Abualsaud <[email protected]>
AuthorDate: Sun Oct 11 21:38:59 2026 -0700
more robust metrics (#40394)
---
.../sdk/io/iceberg/cdc/sink/CommitterMetrics.java | 73 ++++++++--------------
.../beam/sdk/io/iceberg/cdc/sink/package-info.java | 10 +--
.../sdk/io/iceberg/cdc/sink/CommitDeltasTest.java | 34 +++++++---
.../io/iceberg/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 63d2a13c701..c7330386f5d 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 @@ import org.checkerframework.checker.nullness.qual.Nullable;
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<DataFile> dataFiles;
+ final List<DeleteFile> deleteFiles;
final @Nullable Integer firstSpecId;
final Set<Integer> specIds;
final boolean hasEqualityDeletes;
private CommitSummary(
- long dataFileCount,
- long deleteFileCount,
- long dataRecords,
- long equalityDeleteRecords,
- long bytes,
+ List<DataFile> dataFiles,
+ List<DeleteFile> deleteFiles,
@Nullable Integer firstSpecId,
Set<Integer> 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<DataFile> dataFiles, List<DeleteFile>
deleteFiles) {
- long dataRecords = 0;
- long bytes = 0;
@Nullable Integer firstSpecId = null;
Set<Integer> 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 a31833a615e..1793afd3d56 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 @@
* <li><b>snapshotsCreated</b>: 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.
- * <li><b>committedDataFiles</b> / <b>committedDeleteFiles</b>: files
published. Divide by {@code
- * snapshotsCreated} for files per commit. A large ratio points to the
small-file problem
+ * <li><b>committedDataFileRecordCount</b> /
<b>committedEqualityDeleteRecordCount</b>
+ * (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.
+ * <li><b>committedDataFileByteSize</b> /
<b>committedEqualityDeleteByteSize</b> (distributions,
+ * per file): size of each committed file. A low mean points to the
small-file problem
* described below.
- * <li><b>committedRecords</b>: rows committed.
- * <li><b>committedEqualityDeleteRecords</b>: equality-delete rows committed.
- * <li><b>committedBytes</b>: total bytes of committed files.
* <li><b>commitDurationMs</b> (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 95dd70de328..38b38ca1b4d 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.IcebergCatalogConfig;
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 @@ public class CommitDeltasTest {
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 class CommitDeltasTest {
// 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 class CommitDeltasTest {
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 9f8f73f7319..ce89570188f 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 @@ public class CommitterMetricsTest {
.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<DataFile> dataFiles =
+ ImmutableList.of(dataFile(SPEC_0, 10, 100), dataFile(SPEC_0, 20, 200));
+ ImmutableList<DeleteFile> 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 class CommitterMetricsTest {
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 class CommitterMetricsTest {
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());
}
