[Flink]add Flink Writer Metric #2193

wg1026688210 · 2023-10-29T14:43:08Z

add writer metrics:

writeRecordCount
bufferPreemptCount
usedWriteBufferSizeByte
totalWriteBufferSizeByte
flushCostMillis
prepareCommitCostMillis

Test

WriterOperatorTestBase.java
AppendOnlyWriterOperatorTest.java
PrimaryKeyWriterOperatorTest

wg1026688210 · 2023-11-12T15:15:03Z

PTAL @tsreaper @schnappi17 @JingsongLi

tsreaper · 2023-11-13T09:48:39Z

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

+
+        if (writerMetrics != null) {
+            writerMetrics.incWriteRecordNum();
+            writerMetrics.updateWriteCostMS(System.currentTimeMillis() - start);


For most of the time, System.currentTimeMillis() - start will be 0 because records are only buffered in memory. This metrics is only meaningful when we write files to disks or perform compaction.

Also, calling System.currentTimeMillis() for each write will introduce performance issue. Please be very careful with code which is called per record.

tsreaper · 2023-11-13T09:50:48Z

paimon-core/src/main/java/org/apache/paimon/operation/metrics/WriterMetrics.java

+    }
+
+    public void updatePrepareCommitCostMS(long cost) {
+        this.prepareCommitCostMS.update(cost);


NIT: prepareCommitCostMillis or prepareCommitCostMs.

tsreaper · 2023-11-13T09:52:27Z

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

        compactManager
                .getCompactionResult(blocking)
                .ifPresent(
                        result -> {
                            compactBefore.addAll(result.before());
                            compactAfter.addAll(result.after());
                        });
+        writerMetrics.updateSyncLastestCompactionCostMS(System.currentTimeMillis() - start);


If blocking is false this value is certainly 0 because we won't wait for compaction to end. Why do you record this metric?

tsreaper · 2023-11-13T09:54:25Z

paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java

+
+        if (writerMetrics != null) {
+            writerMetrics.incWriteRecordNum();
+            writerMetrics.updateWriteCostMS(System.currentTimeMillis() - start);


Same as above.

tsreaper · 2023-11-13T09:54:55Z

paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java

        Optional<CompactResult> result = compactManager.getCompactionResult(blocking);
        result.ifPresent(this::updateCompactResult);
+        writerMetrics.updateSyncLastestCompactionCostMS(System.currentTimeMillis() - start);


Same as above.

tsreaper · 2023-11-14T04:47:50Z

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java

@@ -125,4 +126,6 @@ List<CommitMessage> prepareCommit(boolean waitCompaction, long commitIdentifier)

    /** With metrics to measure compaction. */
    FileStoreWrite<T> withMetricRegistry(MetricRegistry metricRegistry);
+
+    WriterMetrics getWriterMetrics();


Why do you add this method in the interface when it is only used internally by AbstractFileStoreWrite and its subclasses?

tsreaper · 2023-11-14T04:55:21Z

paimon-core/src/main/java/org/apache/paimon/operation/metrics/WriterMetrics.java

+    public void setMemoryPreemptCount(Supplier<Long> bufferPreemptNumSupplier) {
+        this.stats.bufferPreemptCount = bufferPreemptNumSupplier;
+    }
+
+    public void setUsedWriteBufferSize(Supplier<Long> usedWriteBufferSize) {
+        this.stats.usedWriteBufferSize = usedWriteBufferSize;
+    }
+
+    public void setTotaldWriteBufferSize(Supplier<Long> totaldWriteBufferSize) {
+        this.stats.totalWriteBufferSize = totaldWriteBufferSize;
+    }


Call these methods directly from MemoryPoolFactory and we don't need to introduce the inner Stats class.

tsreaper · 2023-11-14T04:58:14Z

paimon-core/src/main/java/org/apache/paimon/metrics/MetricRegistry.java

+    public MetricGroup tableMetricGroup(String groupName, String tableName, String commitUser) {
        Map<String, String> variables = new LinkedHashMap<>();
        variables.put(KEY_TABLE, tableName);
+        if (commitUser != null) {
+            variables.put(COMMIT_USER, commitUser);
+        }


Why adding commitUser as a metric group variable? For Flink jobs using job id is enough.

I want to associate the Flink job with the commitUser through the tag, making it easier for users to track which Flink job generated the snapshot of the Paimon table. This might be a bit redundant.

tsreaper · 2023-11-16T07:48:13Z

paimon-core/src/main/java/org/apache/paimon/operation/metrics/WriterMetrics.java

+    public void updateBufferFlushCostMS(long bufferFlushCost) {
+        bufferFlushCostMillis.update(bufferFlushCost);
+    }
+
+    public void updatePrepareCommitCostMS(long cost) {
+        this.prepareCommitCostMillis.update(cost);
+    }


You should also change method name to Millis, not MS.

tsreaper · 2023-11-16T08:34:23Z

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java

@@ -77,7 +78,10 @@ public abstract class AbstractFileStoreWrite<T>
    private boolean closeCompactExecutorWhenLeaving = true;
    private boolean ignorePreviousFiles = false;
    protected boolean isStreamingMode = false;
-    private MetricRegistry metricRegistry = null;
+    private MetricRegistry metricRegistry;


Remove useless changes.

tsreaper · 2023-11-16T08:35:17Z

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java

+    @Nullable
+    public WriterMetrics getWriterMetrics() {
+        if (metricRegistry != null && writerMetrics == null) {
+            writerMetrics = new WriterMetrics(metricRegistry, tableName);
+        }
+        return writerMetrics;
+    }


No need for lazy allocation. Just create a new writeMetrics in withMetricRegistry.

tsreaper · 2023-11-16T08:39:08Z

paimon-core/src/main/java/org/apache/paimon/operation/metrics/WriterMetrics.java

+    private static final String GROUP_NAME = "writer";
+
+    private static final int WINDOW_SAMPLE_SIZE = 100;
+    private static final String WRITE_RECORD_NUM = "writeRecordCount";
+
+    private static final String BUFFER_PREEMPT_COUNT = "bufferPreemptCount";
+
+    private static final String USED_WRITE_BUFFER_SIZE = "usedWriteBufferSizeByte";
+
+    private static final String TOTAL_WRITE_BUFFER_SIZE = "totalWriteBufferSizeByte";
+
+    private static final String FLUSH_COST_MILLIS = "flushCostMillis";
+
+    public static final String PREPARE_COMMIT_COST_MILLIS = "prepareCommitCostMillis";
+
+    private final Counter writeRecordNumCounter;
+
+    private final Histogram bufferFlushCostMillis;
+
+    private final Histogram prepareCommitCostMillis;
+
+    private MetricGroup metricGroup;


It seems to me that buffer flushing and prepare commit is related to a certain writer (a certain bucket), while buffer preempt and buffer counts are related to all writers in this subtask. Why do you mix these metrics into one class?

tsreaper · 2023-11-21T03:24:18Z

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java

    private final String tableName;
    private final FileStorePathFactory pathFactory;

+    protected WriterBufferMetric writerBufferMetric;


Is this class member ever used?

tsreaper · 2023-11-21T03:24:45Z

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java

+        Supplier<MemoryPoolFactory> memoryPoolFactorySupplier = writeBufferPoolSupplier();
+        if (metricRegistry != null) {
+            writerBufferMetric =
+                    new WriterBufferMetric(memoryPoolFactorySupplier, metricRegistry, tableName);
+        }


No need for a Supplier. Just override this method in MemoryFileStoreWrite.

tsreaper · 2023-11-21T03:26:14Z

paimon-core/src/main/java/org/apache/paimon/operation/metrics/WriterMetrics.java

+        // cost
+        bufferFlushCostMillis = metricGroup.histogram(FLUSH_COST_MILLIS, WINDOW_SAMPLE_SIZE);
+
+        // prepareCommittime
+        prepareCommitCostMillis =
+                metricGroup.histogram(PREPARE_COMMIT_COST_MILLIS, WINDOW_SAMPLE_SIZE);


I can't quite understand what these comments mean. You should either delete them or add more details.

tsreaper · 2023-11-21T03:27:56Z

paimon-core/src/main/java/org/apache/paimon/operation/metrics/WriterMetrics.java

+    public WriterMetrics(MetricRegistry registry, String tableName, String parition, int bucket) {
+        MetricGroup metricGroup =
+                registry.bucketMetricGroup(GROUP_NAME, tableName, parition, bucket);


Bucket metric groups should be unregistered after the corresponding writers are closed, because a long-running streaming job may modify many many buckets. Add a close method to this class to avoid OOM.

tsreaper · 2023-11-22T03:15:29Z

paimon-core/src/main/java/org/apache/paimon/operation/metrics/WriterBufferMetric.java

+    private MetricGroup metricGroup;
+
+    public WriterBufferMetric(
+            Supplier<MemoryPoolFactory> memoryPoolFactorySupplier,


Why do you need a Supplier? Why not use MemoryPoolFactory directly?

wg1026688210 force-pushed the metric/add_write_metric_v2 branch from 0ea3f1b to 64da853 Compare October 30, 2023 02:31

wg1026688210 changed the title ~~[WIP]add write metric~~ [WIP]add Flink Write Metric Oct 30, 2023

wg1026688210 force-pushed the metric/add_write_metric_v2 branch from 64da853 to c934cac Compare November 12, 2023 15:08

wg1026688210 marked this pull request as ready for review November 12, 2023 15:09

wg1026688210 force-pushed the metric/add_write_metric_v2 branch from c934cac to 776a03e Compare November 12, 2023 15:16

wg1026688210 changed the title ~~[WIP]add Flink Write Metric~~ [Flink]add Flink Writer Metric Nov 13, 2023

tsreaper reviewed Nov 14, 2023

View reviewed changes

wg1026688210 force-pushed the metric/add_write_metric_v2 branch 2 times, most recently from 8f24a21 to a517e95 Compare November 14, 2023 10:51

wg1026688210 requested a review from tsreaper November 14, 2023 12:56

wg1026688210 force-pushed the metric/add_write_metric_v2 branch from ba5a338 to d413373 Compare November 15, 2023 07:37

tsreaper reviewed Nov 16, 2023

View reviewed changes

wg1026688210 force-pushed the metric/add_write_metric_v2 branch 2 times, most recently from c88c5b5 to ca0b6be Compare November 21, 2023 01:40

wg1026688210 requested a review from tsreaper November 21, 2023 01:44

tsreaper reviewed Nov 21, 2023

View reviewed changes

wg1026688210 force-pushed the metric/add_write_metric_v2 branch 3 times, most recently from 042b16a to 4a18b17 Compare November 21, 2023 11:53

wg1026688210 requested a review from tsreaper November 21, 2023 11:55

tsreaper reviewed Nov 22, 2023

View reviewed changes

wg1026688210 added 4 commits November 22, 2023 15:09

writer metric

abf6bc8

writer metric

f7190f7

writer metric

c99a9aa

Adjust the windows size parameter.

502b19f

wg1026688210 force-pushed the metric/add_write_metric_v2 branch 2 times, most recently from a013b93 to 332784f Compare November 22, 2023 08:25

wg1026688210 added 2 commits November 22, 2023 18:07

refactor

503f26b

fix compile error

5de9022

wg1026688210 force-pushed the metric/add_write_metric_v2 branch from 332784f to 5de9022 Compare November 22, 2023 10:08

wg1026688210 requested review from tsreaper November 22, 2023 10:53

tsreaper approved these changes Nov 22, 2023

View reviewed changes

tsreaper merged commit 42b594e into apache:master Nov 22, 2023

wg1026688210 mentioned this pull request Feb 26, 2024

[WIP] add memory buffer preempt count metric #2089

Closed

wg1026688210 mentioned this pull request Apr 8, 2024

[Feature] add writer-buffer preempt metric #1683

Closed

2 tasks

[Flink]add Flink Writer Metric #2193

[Flink]add Flink Writer Metric #2193

Uh oh!

Conversation

wg1026688210 commented Oct 29, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Test

Uh oh!

wg1026688210 commented Nov 12, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wg1026688210 Nov 14, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

wg1026688210 commented Oct 29, 2023 •

edited

Loading

wg1026688210 Nov 14, 2023 •

edited

Loading