diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html index 98c02d7a418fc5..fbf4adcaca1eee 100644 --- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html +++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html @@ -1495,7 +1495,7 @@ @@ -5636,7 +5636,7 @@ @@ -6697,7 +6697,7 @@ diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml index 24ecb76ed2e961..7136406ae8050d 100644 --- a/docs/static/generated/rest_v1_dispatcher.yml +++ b/docs/static/generated/rest_v1_dispatcher.yml @@ -490,7 +490,8 @@ paths: - name: agg in: query description: "Comma-separated list of aggregation modes which should be calculated.\ - \ Available aggregations are: \"min, max, sum, avg, skew\"." + \ Available aggregations are: \"min, max, sum, avg, skew, p50, p90, p99\"\ + ." required: false style: form schema: @@ -1458,7 +1459,8 @@ paths: - name: agg in: query description: "Comma-separated list of aggregation modes which should be calculated.\ - \ Available aggregations are: \"min, max, sum, avg, skew\"." + \ Available aggregations are: \"min, max, sum, avg, skew, p50, p90, p99\"\ + ." required: false style: form schema: @@ -1771,7 +1773,8 @@ paths: - name: agg in: query description: "Comma-separated list of aggregation modes which should be calculated.\ - \ Available aggregations are: \"min, max, sum, avg, skew\"." + \ Available aggregations are: \"min, max, sum, avg, skew, p50, p90, p99\"\ + ." required: false style: form schema: @@ -1899,6 +1902,9 @@ components: - SUM - AVG - SKEW + - P50 + - P90 + - P99 ApplicationDetails: type: object properties: diff --git a/flink-runtime/pom.xml b/flink-runtime/pom.xml index 8a11a818a97656..91598e0f1c2680 100644 --- a/flink-runtime/pom.xml +++ b/flink-runtime/pom.xml @@ -215,6 +215,11 @@ under the License. commons-text + + org.apache.commons + commons-math3 + + commons-cli commons-cli diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java index 79fa9b944ab6da..5b3c36f5c893b5 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AbstractAggregatingMetricsHandler.java @@ -125,7 +125,11 @@ protected CompletableFuture handleRequest( DoubleAccumulator.DoubleAverageFactory averageFactory = null; DoubleAccumulator.DoubleSumFactory sumFactory = null; DoubleAccumulator.DoubleDataSkewFactory skewFactory = null; - // by default we return all aggregations + DoubleAccumulator.DoublePercentileFactory p50Factory = null; + DoubleAccumulator.DoublePercentileFactory p90Factory = null; + DoubleAccumulator.DoublePercentileFactory p99Factory = null; + // by default we return all aggregations (percentiles are opt-in: they + // buffer + sort, so they are only computed when explicitly requested) if (requestedAggregations.isEmpty()) { minimumFactory = DoubleAccumulator.DoubleMinimumFactory.get(); maximumFactory = DoubleAccumulator.DoubleMaximumFactory.get(); @@ -154,6 +158,18 @@ protected CompletableFuture handleRequest( case SKEW: skewFactory = DoubleAccumulator.DoubleDataSkewFactory.get(); break; + case P50: + p50Factory = + DoubleAccumulator.DoublePercentileFactory.p50(); + break; + case P90: + p90Factory = + DoubleAccumulator.DoublePercentileFactory.p90(); + break; + case P99: + p99Factory = + DoubleAccumulator.DoublePercentileFactory.p99(); + break; default: log.warn( "Unsupported aggregation specified: {}", @@ -167,7 +183,10 @@ protected CompletableFuture handleRequest( maximumFactory, averageFactory, sumFactory, - skewFactory); + skewFactory, + p50Factory, + p90Factory, + p99Factory); return getAggregatedMetricValues( stores, requestedMetrics, metricAccumulatorFactory); @@ -255,18 +274,27 @@ private static class MetricAccumulatorFactory { @Nullable private final DoubleAccumulator.DoubleSumFactory sumFactory; @Nullable private final DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory; + @Nullable private final DoubleAccumulator.DoublePercentileFactory p50Factory; + @Nullable private final DoubleAccumulator.DoublePercentileFactory p90Factory; + @Nullable private final DoubleAccumulator.DoublePercentileFactory p99Factory; private MetricAccumulatorFactory( @Nullable DoubleAccumulator.DoubleMinimumFactory minimumFactory, @Nullable DoubleAccumulator.DoubleMaximumFactory maximumFactory, @Nullable DoubleAccumulator.DoubleAverageFactory averageFactory, @Nullable DoubleAccumulator.DoubleSumFactory sumFactory, - @Nullable DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory) { + @Nullable DoubleAccumulator.DoubleDataSkewFactory dataSkewFactory, + @Nullable DoubleAccumulator.DoublePercentileFactory p50Factory, + @Nullable DoubleAccumulator.DoublePercentileFactory p90Factory, + @Nullable DoubleAccumulator.DoublePercentileFactory p99Factory) { this.minimumFactory = minimumFactory; this.maximumFactory = maximumFactory; this.averageFactory = averageFactory; this.sumFactory = sumFactory; this.dataSkewFactory = dataSkewFactory; + this.p50Factory = p50Factory; + this.p90Factory = p90Factory; + this.p99Factory = p99Factory; } MetricAccumulator get(String metricName, double init) { @@ -276,7 +304,10 @@ MetricAccumulator get(String metricName, double init) { maximumFactory == null ? null : maximumFactory.get(init), averageFactory == null ? null : averageFactory.get(init), sumFactory == null ? null : sumFactory.get(init), - dataSkewFactory == null ? null : dataSkewFactory.get(init)); + dataSkewFactory == null ? null : dataSkewFactory.get(init), + p50Factory == null ? null : p50Factory.get(init), + p90Factory == null ? null : p90Factory.get(init), + p99Factory == null ? null : p99Factory.get(init)); } } @@ -288,6 +319,9 @@ private static class MetricAccumulator { @Nullable private final DoubleAccumulator avg; @Nullable private final DoubleAccumulator sum; @Nullable private final DoubleAccumulator skew; + @Nullable private final DoubleAccumulator p50; + @Nullable private final DoubleAccumulator p90; + @Nullable private final DoubleAccumulator p99; private MetricAccumulator( String metricName, @@ -295,13 +329,19 @@ private MetricAccumulator( @Nullable DoubleAccumulator max, @Nullable DoubleAccumulator avg, @Nullable DoubleAccumulator sum, - @Nullable DoubleAccumulator.DoubleDataSkew skew) { + @Nullable DoubleAccumulator.DoubleDataSkew skew, + @Nullable DoubleAccumulator p50, + @Nullable DoubleAccumulator p90, + @Nullable DoubleAccumulator p99) { this.metricName = Preconditions.checkNotNull(metricName); this.min = min; this.max = max; this.avg = avg; this.sum = sum; this.skew = skew; + this.p50 = p50; + this.p90 = p90; + this.p99 = p99; } void add(double value) { @@ -320,6 +360,15 @@ void add(double value) { if (skew != null) { skew.add(value); } + if (p50 != null) { + p50.add(value); + } + if (p90 != null) { + p90.add(value); + } + if (p99 != null) { + p99.add(value); + } } AggregatedMetric get() { @@ -329,7 +378,10 @@ AggregatedMetric get() { max == null ? null : max.getValue(), avg == null ? null : avg.getValue(), sum == null ? null : sum.getValue(), - skew == null ? null : skew.getValue()); + skew == null ? null : skew.getValue(), + p50 == null ? null : p50.getValue(), + p90 == null ? null : p90.getValue(), + p99 == null ? null : p99.getValue()); } } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulator.java index 7dd9ef1e386655..6e12e16076dcdb 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulator.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulator.java @@ -20,6 +20,9 @@ import org.apache.flink.annotation.VisibleForTesting; +import org.apache.commons.math3.stat.descriptive.rank.Percentile; +import org.apache.commons.math3.stat.ranking.NaNStrategy; + import java.util.ArrayList; import java.util.List; @@ -144,6 +147,36 @@ public static DoubleDataSkewFactory get() { } } + /** Factory for {@link DoublePercentile}. */ + final class DoublePercentileFactory implements DoubleAccumulatorFactory { + private static final DoublePercentileFactory P50 = new DoublePercentileFactory(50); + private static final DoublePercentileFactory P90 = new DoublePercentileFactory(90); + private static final DoublePercentileFactory P99 = new DoublePercentileFactory(99); + + private final double percentile; + + private DoublePercentileFactory(double percentile) { + this.percentile = percentile; + } + + @Override + public DoublePercentile get(double init) { + return new DoublePercentile(percentile, init); + } + + public static DoublePercentileFactory p50() { + return P50; + } + + public static DoublePercentileFactory p90() { + return P90; + } + + public static DoublePercentileFactory p99() { + return P99; + } + } + /** {@link DoubleAccumulator} that returns the maximum value. */ final class DoubleMaximum implements DoubleAccumulator { @@ -302,4 +335,40 @@ public String getName() { return NAME; } } + + /** + * {@link DoubleAccumulator} that returns a percentile (e.g. p50/p90/p99) over all values. A + * percentile needs the whole sample, so this buffers every value (like {@link DoubleDataSkew}) + * and computes the result lazily in {@link #getValue()}. + */ + final class DoublePercentile implements DoubleAccumulator { + + private final double percentile; + + private final String name; + + private final List values = new ArrayList<>(); + + private DoublePercentile(double percentile, double init) { + this.percentile = percentile; + this.name = "p" + (int) percentile; + values.add(init); + } + + @Override + public void add(double value) { + values.add(value); + } + + @Override + public double getValue() { + double[] arr = values.stream().mapToDouble(Double::doubleValue).toArray(); + return new Percentile().withNaNStrategy(NaNStrategy.FIXED).evaluate(arr, percentile); + } + + @Override + public String getName() { + return name; + } + } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedMetric.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedMetric.java index dbeb432198142c..4ca4c5d1642c5e 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedMetric.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/AggregatedMetric.java @@ -45,6 +45,12 @@ public class AggregatedMetric { private static final String FIELD_NAME_SKEW = "skew"; + private static final String FIELD_NAME_P50 = "p50"; + + private static final String FIELD_NAME_P90 = "p90"; + + private static final String FIELD_NAME_P99 = "p99"; + @JsonProperty(value = FIELD_NAME_ID, required = true) private final String id; @@ -68,6 +74,18 @@ public class AggregatedMetric { @JsonProperty(FIELD_NAME_SKEW) private final Double skew; + @JsonInclude(JsonInclude.Include.NON_NULL) + @JsonProperty(FIELD_NAME_P50) + private final Double p50; + + @JsonInclude(JsonInclude.Include.NON_NULL) + @JsonProperty(FIELD_NAME_P90) + private final Double p90; + + @JsonInclude(JsonInclude.Include.NON_NULL) + @JsonProperty(FIELD_NAME_P99) + private final Double p99; + @JsonCreator public AggregatedMetric( final @JsonProperty(value = FIELD_NAME_ID, required = true) String id, @@ -75,7 +93,10 @@ public AggregatedMetric( final @Nullable @JsonProperty(FIELD_NAME_MAX) Double max, final @Nullable @JsonProperty(FIELD_NAME_AVG) Double avg, final @Nullable @JsonProperty(FIELD_NAME_SUM) Double sum, - final @Nullable @JsonProperty(FIELD_NAME_SKEW) Double skew) { + final @Nullable @JsonProperty(FIELD_NAME_SKEW) Double skew, + final @Nullable @JsonProperty(FIELD_NAME_P50) Double p50, + final @Nullable @JsonProperty(FIELD_NAME_P90) Double p90, + final @Nullable @JsonProperty(FIELD_NAME_P99) Double p99) { this.id = requireNonNull(id, "id must not be null"); this.min = min; @@ -83,10 +104,13 @@ public AggregatedMetric( this.avg = avg; this.sum = sum; this.skew = skew; + this.p50 = p50; + this.p90 = p90; + this.p99 = p99; } public AggregatedMetric(final @JsonProperty(value = FIELD_NAME_ID, required = true) String id) { - this(id, null, null, null, null, null); + this(id, null, null, null, null, null, null, null, null); } @JsonIgnore @@ -119,6 +143,21 @@ public Double getSkew() { return skew; } + @JsonIgnore + public Double getP50() { + return p50; + } + + @JsonIgnore + public Double getP90() { + return p90; + } + + @JsonIgnore + public Double getP99() { + return p99; + } + @Override public String toString() { return "AggregatedMetric{" @@ -140,6 +179,15 @@ public String toString() { + ", skew='" + skew + '\'' + + ", p50='" + + p50 + + '\'' + + ", p90='" + + p90 + + '\'' + + ", p99='" + + p99 + + '\'' + '}'; } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsAggregationParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsAggregationParameter.java index 2f1cfcd867a13d..378a9351bdd64d 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsAggregationParameter.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/MetricsAggregationParameter.java @@ -61,5 +61,8 @@ public enum AggregationMode { SUM, AVG, SKEW, + P50, + P90, + P99, } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingMetricsHandlerTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingMetricsHandlerTestBase.java index 05ec56e48e738d..18d5fbdfed3ff8 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingMetricsHandlerTestBase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingMetricsHandlerTestBase.java @@ -385,6 +385,9 @@ void testDefaultAggregation() throws Exception { assertThat(aggregatedMetric.getAvg()).isCloseTo(2.0, within(0.1)); assertThat(aggregatedMetric.getSum()).isCloseTo(4.0, within(0.1)); assertThat(aggregatedMetric.getSkew()).isCloseTo(50.0, within(0.1)); + assertThat(aggregatedMetric.getP50()).isNull(); + assertThat(aggregatedMetric.getP90()).isNull(); + assertThat(aggregatedMetric.getP99()).isNull(); } @Test @@ -419,4 +422,42 @@ void testDataSkewAggregation() throws Exception { assertThat(aggregatedMetric.getMax()).isNull(); assertThat(aggregatedMetric.getSum()).isNull(); } + + @Test + void testPercentileAggregation() throws Exception { + Map> queryParams = new HashMap<>(4); + queryParams.put("get", Collections.singletonList("abc.metric1")); + queryParams.put("agg", Arrays.asList("p50", "p90", "p99")); + + HandlerRequest request = + HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + handler.getMessageHeaders().getUnresolvedMessageParameters(), + pathParameters, + queryParams, + Collections.emptyList()); + + AggregatedMetricsResponseBody response = + handler.handleRequest(request, MOCK_DISPATCHER_GATEWAY).get(); + + Collection aggregatedMetrics = response.getMetrics(); + + assertThat(aggregatedMetrics).hasSize(1); + AggregatedMetric aggregatedMetric = aggregatedMetrics.iterator().next(); + + assertThat(aggregatedMetric.getId()).isEqualTo("abc.metric1"); + // abc.metric1 has the data points [1, 3]; commons-math3 LEGACY percentile + // (pos = percentile * (N + 1) / 100, N = 2): + // p50 -> pos = 1.5 -> 1 + 0.5 * (3 - 1) = 2 + // p90 -> pos = 2.7 -> pos >= N, clamped to max = 3 + // p99 -> pos = 2.97 -> pos >= N, clamped to max = 3 + assertThat(aggregatedMetric.getP50()).isCloseTo(2.0, within(0.1)); + assertThat(aggregatedMetric.getP90()).isCloseTo(3.0, within(0.1)); + assertThat(aggregatedMetric.getP99()).isCloseTo(3.0, within(0.1)); + assertThat(aggregatedMetric.getMin()).isNull(); + assertThat(aggregatedMetric.getMax()).isNull(); + assertThat(aggregatedMetric.getAvg()).isNull(); + assertThat(aggregatedMetric.getSum()).isNull(); + assertThat(aggregatedMetric.getSkew()).isNull(); + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulatorTest.java index cdb9b2742e18a9..b033606aae6c8f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/DoubleAccumulatorTest.java @@ -53,6 +53,47 @@ public void testDataSkewOnSingleValueList() { assertThat(dataSkew.getValue()).isEqualTo(0.0); } + @Test + public void testPercentile() { + // Sample: 10, 20, ..., 100 (N=10). commons-math3 LEGACY interpolation uses + // pos = percentile * (N + 1) / 100: + // p50 -> pos = 5.5 -> 50 + 0.5 * (60 - 50) = 55 + // p90 -> pos = 9.9 -> 90 + 0.9 * (100 - 90) = 99 + // p99 -> pos = 10.89 -> pos >= N, clamped to max = 100 + assertThat(percentileOf(DoubleAccumulator.DoublePercentileFactory.p50()).getValue()) + .isCloseTo(55.0, within(0.001)); + assertThat(percentileOf(DoubleAccumulator.DoublePercentileFactory.p90()).getValue()) + .isCloseTo(99.0, within(0.001)); + assertThat(percentileOf(DoubleAccumulator.DoublePercentileFactory.p99()).getValue()) + .isCloseTo(100.0, within(0.001)); + } + + @Test + public void testPercentileOnSingleValueList() { + DoubleAccumulator.DoublePercentile percentile = + DoubleAccumulator.DoublePercentileFactory.p90().get(42.0); + assertThat(percentile.getValue()).isEqualTo(42.0); + } + + @Test + public void testPercentileName() { + assertThat(DoubleAccumulator.DoublePercentileFactory.p50().get(1.0).getName()) + .isEqualTo("p50"); + assertThat(DoubleAccumulator.DoublePercentileFactory.p90().get(1.0).getName()) + .isEqualTo("p90"); + assertThat(DoubleAccumulator.DoublePercentileFactory.p99().get(1.0).getName()) + .isEqualTo("p99"); + } + + private static DoubleAccumulator.DoublePercentile percentileOf( + DoubleAccumulator.DoublePercentileFactory factory) { + DoubleAccumulator.DoublePercentile percentile = factory.get(10.0); + for (double value : new double[] {20, 30, 40, 50, 60, 70, 80, 90, 100}) { + percentile.add(value); + } + return percentile; + } + private static Stream dataSkewTests() { // Data set, followed by the expected data skew percentage return Stream.of(