[metrics] Add io-metrics for paimon in Flink #5550
Conversation
|
1、You add lots of metric, I suggest you sent the discuss email to [email protected] first. |
Roger that. |
|
|
||
| <dependency> | ||
| <groupId>org.apache.commons</groupId> | ||
| <artifactId>commons-math3</artifactId> |
There was a problem hiding this comment.
Please remove this dep
I moved some metrics classes from paimon-core to paimon-common project, and the class DescriptiveStatisticsHistogram depends on this.
| * | ||
| * <p>It allows users to monitor and track file I/O operations (e.g., read, write, delete, rename). | ||
| */ | ||
| public class FileIOWrapper implements FileIO { |
| public static final String GROUP_NAME = "source"; | ||
| private final MetricGroup metricGroup; | ||
|
|
||
| private final AtomicLong readBytes = new AtomicLong(0); |
There was a problem hiding this comment.
only keep:
String READ_BYTES = "read.bytes";
String READ_OPERATIONS = "read.operations";
String WRITE_BYTES = "write.bytes";
String WRITE_OPERATIONS = "write.operations";
There was a problem hiding this comment.
only keep:
String READ_BYTES = "read.bytes"; String READ_OPERATIONS = "read.operations"; String WRITE_BYTES = "write.bytes"; String WRITE_OPERATIONS = "write.operations";
Other metrics have been removed.
| TableSchema tableSchema, | ||
| CatalogEnvironment catalogEnvironment) { | ||
| this.fileIO = fileIO; | ||
| this.fileIO = new FileIOWrapper(fileIO); |
There was a problem hiding this comment.
We may need to introduce an option to enable MetricsFileIO.
There was a problem hiding this comment.
Done
We may need to introduce an option to enable
MetricsFileIO.
|
I'v done refactor for making with MetricRegistry public in #5578 |
rebase on this && add uts & all checks passed, cc~ @JingsongLi |
JingsongLi
left a comment
JingsongLi
left a comment
There was a problem hiding this comment.
Please only use TableWrite.withMetricRegistry and TableRead.withMetricRegistry, do not introduce any API for metrics.
And please make sure only io metric enabled, the file io can be MetricsFileIO.
| @Override | ||
| public SeekableInputStream newInputStream(Path path) throws IOException { | ||
| SeekableInputStream inputStream = fileIO.newInputStream(path); | ||
| return new SeekableInputStreamIOWrapper(inputStream, this.inputMetrics); |
There was a problem hiding this comment.
Please consider SeekableInputStream can be VectoredReadable.
There was a problem hiding this comment.
Please consider
SeekableInputStreamcan beVectoredReadable.
done.
| import java.util.concurrent.atomic.AtomicLong; | ||
|
|
||
| /** Collects and monitors outp stream metrics. */ | ||
| public class OutputMetrics { |
There was a problem hiding this comment.
Please merge these tow classes into one.
There was a problem hiding this comment.
Please merge these tow classes into one.
done.
|
|
||
| /** Collects and monitors outp stream metrics. */ | ||
| public class OutputMetrics { | ||
| public static final String GROUP_NAME = "sink"; |
There was a problem hiding this comment.
group name should be io.
There was a problem hiding this comment.
group name should be
io.
done.
If we only use TableRead.withMetricRegistry, it would require too may modifications. However, in reality, we're always using and modifying the same AbstractFileStoreTable fileio reference. I've attempted to register metrics directly to the AbstractFileStoreTable within the ReadBuilder. This way, all subsequent operations like new scan, new streamscan, or new read can directly reuse the metrics collected by the AbstractFileStoreTable. |
zhuanshenbsj1
force-pushed
the
metrics
branch
4 times, most recently
from
f2e8fe8 to
99e93e4
Compare
zhuanshenbsj1
force-pushed
the
metrics
branch
4 times, most recently
from
20eabb5 to
2b3fd4c
Compare
|
@JingsongLi Could you please take a look at this PR again for any remaining issues， thank ~ |
3 participants
Purpose
Tests
API and Format
Documentation