You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "mxm (via GitHub)" <gi...@apache.org> on 2023/03/28 18:36:57 UTC
[GitHub] [flink-kubernetes-operator] mxm opened a new pull request, #556: [FLINK-30575] Set processing capacity to infinite if task is idle
mxm opened a new pull request, #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556
In a prior change we set the processing rates and capacity (true processing rate) to near zero values to allow scaling down in case of idle tasks. This has worked well but caused issues for tasks which are fed via upstream selective outputs (side outputs). Those tests usually received none or only a few records of the upstream outputted records. Setting low values for the incoming records rate will yield a relatively low true processing rate which quickly triggers upscaling even if no records are being processed at all.
We don't currently have per-output/per-input metrics, so the best we can do, is to assume infinite processing capacity if we don't receive any records or are completely idle. Once we received records, we can measure our true processing rate. In the case of selective outputs, we will still assume we receive all the data which is suboptimal but far better than scaling up all the way to the max parallelism as we have seen for the near zero true processing rates during idleness.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] mxm merged pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm merged PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1151617424
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -38,10 +38,12 @@ public class ScalingMetrics {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetrics.class);
/**
- * The minimum value to avoid using zero values which cause side effects like division by zero
- * or out of bounds (infinitive) floats.
+ * A large, effectively infinite value which continues to allow arithmetics for the scaling
+ * algorithm. It is used in case the task is not processing data or is completely idle. We can't
+ * accurately sample its true processing rate in this case and assume a high capacity until we
+ * know any better. This is to allow scale down in case a task becomes completely idle.
*/
- public static final double EFFECTIVELY_ZERO = 1e-10;
+ public static final double EFFECTIVELY_INFINITE = 1e100;
Review Comment:
Any reason for not actually using `Double.POSITIVE_INFINITY`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1151733955
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -38,10 +38,12 @@ public class ScalingMetrics {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetrics.class);
/**
- * The minimum value to avoid using zero values which cause side effects like division by zero
- * or out of bounds (infinitive) floats.
+ * A large, effectively infinite value which continues to allow arithmetics for the scaling
+ * algorithm. It is used in case the task is not processing data or is completely idle. We can't
+ * accurately sample its true processing rate in this case and assume a high capacity until we
+ * know any better. This is to allow scale down in case a task becomes completely idle.
*/
- public static final double EFFECTIVELY_ZERO = 1e-10;
+ public static final double EFFECTIVELY_INFINITE = 1e100;
Review Comment:
Let's use whatever makes sense and leads to the correct metrics, I was just curious
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] mxm commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1151779136
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -38,10 +38,12 @@ public class ScalingMetrics {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetrics.class);
/**
- * The minimum value to avoid using zero values which cause side effects like division by zero
- * or out of bounds (infinitive) floats.
+ * A large, effectively infinite value which continues to allow arithmetics for the scaling
+ * algorithm. It is used in case the task is not processing data or is completely idle. We can't
+ * accurately sample its true processing rate in this case and assume a high capacity until we
+ * know any better. This is to allow scale down in case a task becomes completely idle.
*/
- public static final double EFFECTIVELY_ZERO = 1e-10;
+ public static final double EFFECTIVELY_INFINITE = 1e100;
Review Comment:
Updated pushed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] mxm commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1153480322
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -251,12 +251,20 @@ private static double getAverage(
ScalingMetric metric,
JobVertexID jobVertexId,
SortedMap<Instant, Map<JobVertexID, Map<ScalingMetric, Double>>> metricsHistory) {
- return StatUtils.mean(
+ double[] metricsValues =
metricsHistory.values().stream()
.map(m -> m.get(jobVertexId))
.filter(m -> m.containsKey(metric))
.mapToDouble(m -> m.get(metric))
.filter(d -> !Double.isNaN(d))
- .toArray());
+ .toArray();
+ for (double metricsValue : metricsValues) {
+ if (Double.isInfinite(metricsValue)) {
+ // As long as infinite values are present, we can't properly average. We need to
+ // wait until they are evicted.
+ return metricsValue;
Review Comment:
Changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] mxm commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1153487988
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -251,12 +251,20 @@ private static double getAverage(
ScalingMetric metric,
JobVertexID jobVertexId,
SortedMap<Instant, Map<JobVertexID, Map<ScalingMetric, Double>>> metricsHistory) {
- return StatUtils.mean(
+ double[] metricsValues =
metricsHistory.values().stream()
.map(m -> m.get(jobVertexId))
.filter(m -> m.containsKey(metric))
.mapToDouble(m -> m.get(metric))
.filter(d -> !Double.isNaN(d))
- .toArray());
+ .toArray();
+ for (double metricsValue : metricsValues) {
+ if (Double.isInfinite(metricsValue)) {
+ // As long as infinite values are present, we can't properly average. We need to
+ // wait until they are evicted.
+ return metricsValue;
Review Comment:
Actually, reverted. I'm not sure whether we should return positive infinity here. We could potentially have negative infinity values which would be removed here. Would just keep things as-is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] mxm commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1151704447
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -38,10 +38,12 @@ public class ScalingMetrics {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetrics.class);
/**
- * The minimum value to avoid using zero values which cause side effects like division by zero
- * or out of bounds (infinitive) floats.
+ * A large, effectively infinite value which continues to allow arithmetics for the scaling
+ * algorithm. It is used in case the task is not processing data or is completely idle. We can't
+ * accurately sample its true processing rate in this case and assume a high capacity until we
+ * know any better. This is to allow scale down in case a task becomes completely idle.
*/
- public static final double EFFECTIVELY_ZERO = 1e-10;
+ public static final double EFFECTIVELY_INFINITE = 1e100;
Review Comment:
That's not a real number. Arithmetics on positive infinity can result in Nan. For example:
```
> Double.POSITIVE_INFINITY * 0
NaN
```
But point taken, will push an update in a sec.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] mxm commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1151779136
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -38,10 +38,12 @@ public class ScalingMetrics {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetrics.class);
/**
- * The minimum value to avoid using zero values which cause side effects like division by zero
- * or out of bounds (infinitive) floats.
+ * A large, effectively infinite value which continues to allow arithmetics for the scaling
+ * algorithm. It is used in case the task is not processing data or is completely idle. We can't
+ * accurately sample its true processing rate in this case and assume a high capacity until we
+ * know any better. This is to allow scale down in case a task becomes completely idle.
*/
- public static final double EFFECTIVELY_ZERO = 1e-10;
+ public static final double EFFECTIVELY_INFINITE = 1e100;
Review Comment:
Update pushed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #556: [FLINK-30575] Set processing capacity to infinite if task is idle
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #556:
URL: https://github.com/apache/flink-kubernetes-operator/pull/556#discussion_r1151818091
##########
flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingMetricEvaluator.java:
##########
@@ -251,12 +251,20 @@ private static double getAverage(
ScalingMetric metric,
JobVertexID jobVertexId,
SortedMap<Instant, Map<JobVertexID, Map<ScalingMetric, Double>>> metricsHistory) {
- return StatUtils.mean(
+ double[] metricsValues =
metricsHistory.values().stream()
.map(m -> m.get(jobVertexId))
.filter(m -> m.containsKey(metric))
.mapToDouble(m -> m.get(metric))
.filter(d -> !Double.isNaN(d))
- .toArray());
+ .toArray();
+ for (double metricsValue : metricsValues) {
+ if (Double.isInfinite(metricsValue)) {
+ // As long as infinite values are present, we can't properly average. We need to
+ // wait until they are evicted.
+ return metricsValue;
Review Comment:
This confused me a little, should we simply return Double.positive_infinity here explicitly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org