You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Naireen (via GitHub)" <gi...@apache.org> on 2024/07/30 22:05:34 UTC

[PR] Filter empty distribution metrics [beam]

Naireen opened a new pull request, #32027:
URL: https://github.com/apache/beam/pull/32027

   Filter empty distributions on the python side so its not sent to the runner for processing, since it adds necessary overhead. This can occur when processing failed, or there was an empty split created (for batch pipelines) 
   ------------------------
   
   Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
   
    - [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
    - [ ] Update `CHANGES.md` with noteworthy changes.
    - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
   
   See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
   
   To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
   
   GitHub Actions Tests Status (on master branch)
   ------------------------------------------------------------------------------------------------
   [![Build python source distribution and wheels](https://github.com/apache/beam/workflows/Build%20python%20source%20distribution%20and%20wheels/badge.svg?branch=master&event=schedule)](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
   [![Python tests](https://github.com/apache/beam/workflows/Python%20tests/badge.svg?branch=master&event=schedule)](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
   [![Java tests](https://github.com/apache/beam/workflows/Java%20Tests/badge.svg?branch=master&event=schedule)](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
   [![Go tests](https://github.com/apache/beam/workflows/Go%20tests/badge.svg?branch=master&event=schedule)](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
   
   See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
   


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1714416927


##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -213,11 +216,18 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     metric: The DistributionData for the metric.
     ptransform: The ptransform id used as a label.
   """
-  labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
-  payload = _encode_distribution(
-      coders.VarIntCoder(), metric.count, metric.sum, metric.min, metric.max)
-  return create_monitoring_info(
-      USER_DISTRIBUTION_URN, DISTRIBUTION_INT64_TYPE, payload, labels)
+  if metric.count <= 0:

Review Comment:
   nit: I'd prefer to lead the condition with the happy path (`if metric.count > 0`)



##########
sdks/python/apache_beam/runners/worker/operations.py:
##########
@@ -605,18 +605,23 @@ def pcollection_count_monitoring_infos(self, tag_to_pcollection_id):
         receiver.opcounter.element_counter.value(),
         pcollection=pcollection_id,
     )
+    all_monitoring_infos[monitoring_infos.to_key(elem_count_mi)] = elem_count_mi

Review Comment:
   Do we need to check that `elem_count_mi` is not `None` here?



##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -213,11 +216,18 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     metric: The DistributionData for the metric.
     ptransform: The ptransform id used as a label.
   """
-  labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
-  payload = _encode_distribution(
-      coders.VarIntCoder(), metric.count, metric.sum, metric.min, metric.max)
-  return create_monitoring_info(
-      USER_DISTRIBUTION_URN, DISTRIBUTION_INT64_TYPE, payload, labels)
+  if metric.count <= 0:
+    _LOGGER.debug(
+        'Expected a non zero distribution count for %s metric but received %s' %
+        (metric.name, metric.count))
+    return

Review Comment:
   Since we return `None` (implicitly), we need to change the method typehint to `# type: (...) -> Optional[metrics_pb2.MonitoringInfo]`. 



##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -214,6 +214,11 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     ptransform: The ptransform id used as a label.
   """
   labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
+  if metric.count <= 0:
+    raise TypeError(

Review Comment:
   I think not emitting sounds fine, left some comments.



##########
sdks/python/apache_beam/runners/worker/operations.py:
##########
@@ -605,18 +605,23 @@ def pcollection_count_monitoring_infos(self, tag_to_pcollection_id):
         receiver.opcounter.element_counter.value(),
         pcollection=pcollection_id,
     )
+    all_monitoring_infos[monitoring_infos.to_key(elem_count_mi)] = elem_count_mi

Review Comment:
   Also, what happens with distribution if the count is 0 after all? For example , pipeline reads from a file, but there are no elements, and pipeline stops. will this case be handled correctly?



##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -213,11 +216,18 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     metric: The DistributionData for the metric.
     ptransform: The ptransform id used as a label.
   """
-  labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
-  payload = _encode_distribution(
-      coders.VarIntCoder(), metric.count, metric.sum, metric.min, metric.max)
-  return create_monitoring_info(
-      USER_DISTRIBUTION_URN, DISTRIBUTION_INT64_TYPE, payload, labels)
+  if metric.count <= 0:
+    _LOGGER.debug(
+        'Expected a non zero distribution count for %s metric but received %s' %

Review Comment:
   In which situation this log message will be actionable? I wonder if we should remove this log if it commonly happens (e.g. retries). 



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2285008106

   So with the current approach of returning None, this fails:
   
    ./gradlew :sdks:python:test-suites:tox:pycommon:mypy (part of the Precommig Python Lint test)
    
    Essentially the issue is that we have a dict of strings input, which should produce a dict of Monitoring info. Since we aren't raising errors, pyhint fails since it now expects `monitoring_infos.int64_distribution` changes from  metrics_pb2.MonitoringInfo to Optional[metrics_pb2.MonitoringInfo], which makes operations.pcollection_count_monitoring_infos invalid, can't go from Dict[FrozenSet, metrics_pb2.MonitoringInfo] to Dict[FrozenSet, Optional[metrics_pb2.MonitoringInfo]]
    
    Ideally we do some filtering here, if we went down this approach, which isn't the most performant
    
    The other option was to raise an error, gaurd the runner against it, so then if a user creates a invalid distribution, it would error (which is a breaking change, which also isn't desirable)
    
    Then which of the two approaches make sense? Is there a third option? 
   
   


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2259437042

    Run Python_Coverage PreCommit


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1714021067


##########
sdks/python/apache_beam/runners/worker/operations.py:
##########
@@ -605,18 +605,23 @@ def pcollection_count_monitoring_infos(self, tag_to_pcollection_id):
         receiver.opcounter.element_counter.value(),
         pcollection=pcollection_id,
     )
-
-    (unused_mean, sum, count, min, max) = (
-        receiver.opcounter.mean_byte_counter.value())
-
-    sampled_byte_count = monitoring_infos.int64_distribution(
-        monitoring_infos.SAMPLED_BYTE_SIZE_URN,
-        DistributionData(sum, count, min, max),
-        pcollection=pcollection_id,
-    )
     all_monitoring_infos[monitoring_infos.to_key(elem_count_mi)] = elem_count_mi
-    all_monitoring_infos[monitoring_infos.to_key(
-        sampled_byte_count)] = sampled_byte_count
+
+    try:

Review Comment:
   Done



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2299287785

   Discussed offline


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1711831823


##########
sdks/python/apache_beam/runners/worker/operations.py:
##########
@@ -605,18 +605,23 @@ def pcollection_count_monitoring_infos(self, tag_to_pcollection_id):
         receiver.opcounter.element_counter.value(),
         pcollection=pcollection_id,
     )
-
-    (unused_mean, sum, count, min, max) = (
-        receiver.opcounter.mean_byte_counter.value())
-
-    sampled_byte_count = monitoring_infos.int64_distribution(
-        monitoring_infos.SAMPLED_BYTE_SIZE_URN,
-        DistributionData(sum, count, min, max),
-        pcollection=pcollection_id,
-    )
     all_monitoring_infos[monitoring_infos.to_key(elem_count_mi)] = elem_count_mi
-    all_monitoring_infos[monitoring_infos.to_key(
-        sampled_byte_count)] = sampled_byte_count
+
+    try:

Review Comment:
   would it make sense to check here `if(count > 0)` instead?



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2259430612

   Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
   
   R: @tvalentyn for label python.
   
   Available commands:
   - `stop reviewer notifications` - opt out of the automated review tooling
   - `remind me after tests pass` - tag the comment author after tests pass
   - `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
   
   The PR bot will only process comments in the main thread (not review comments).


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1711836265


##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -214,6 +214,11 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     ptransform: The ptransform id used as a label.
   """
   labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
+  if metric.count <= 0:
+    raise TypeError(

Review Comment:
   I am not seeing where this exception will be caught, hence same question - can we avoid entering this codepath instead of catching the exception in a try-catch.
   
   ValueError would be more appropriate here.



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #32027: Filter empty distribution metrics
URL: https://github.com/apache/beam/pull/32027


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1711836741


##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -214,6 +214,11 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     ptransform: The ptransform id used as a label.
   """
   labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
+  if metric.count <= 0:
+    raise TypeError(
+        'Expected a non zero distribution count for %s metric but received %s' %
+        (metric, metric.count))

Review Comment:
   you probably meant metric.name here



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2441492704

   This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1699152741


##########
sdks/python/apache_beam/runners/worker/bundle_processor.py:
##########
@@ -1257,7 +1257,18 @@ def monitoring_infos(self):
       all_monitoring_infos_dict.update(
           op.monitoring_infos(transform_id, dict(tag_to_pcollection_id)))
 
-    return list(all_monitoring_infos_dict.values())
+    # Filter distributions that have values of zero counts.
+    filtered_dict = {}

Review Comment:
   A few comments
   
   1) The monitoring_info collection is a performance-sensitive part of Beam, so I am vary of adding additional processing here
   2) This change will affect only new SDKs, runners would still have to accommodate the behavior for older SDKs.
   
   3) Is it possible to not emit the empty distributions when they are initially created, so that they don't appear in all_monitoring_infos_dict in the first place?



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1700831351


##########
sdks/python/apache_beam/runners/worker/bundle_processor.py:
##########
@@ -1257,7 +1257,18 @@ def monitoring_infos(self):
       all_monitoring_infos_dict.update(
           op.monitoring_infos(transform_id, dict(tag_to_pcollection_id)))
 
-    return list(all_monitoring_infos_dict.values())
+    # Filter distributions that have values of zero counts.
+    filtered_dict = {}

Review Comment:
   Done, did it so if the distribution creation returns an error, it never is added to all_monitoring_infos_dict



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2424913605

   This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1711841887


##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -214,6 +214,11 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     ptransform: The ptransform id used as a label.
   """
   labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
+  if metric.count <= 0:
+    raise TypeError(

Review Comment:
   Also if exception is uncaught, it can create a breaking change for users, so as much as possible I'd prefer to not enter the exception path or fail silently 



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "Naireen (via GitHub)" <gi...@apache.org>.
Naireen commented on code in PR #32027:
URL: https://github.com/apache/beam/pull/32027#discussion_r1714027688


##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -214,6 +214,11 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     ptransform: The ptransform id used as a label.
   """
   labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
+  if metric.count <= 0:
+    raise TypeError(

Review Comment:
   For the Dataflow Runner, there are checks when we create the distribution, so that should be fine. For a user defined counter, what behaviour do you want? I agree we shouldn't introduce breaking changes. Is it fine to not emit a counter in that case? If we still want to emit something here, then we'll have an empty counter, and we'd have to filter it out in the runner to prevent sending it to the backend (which we don't want to add either based on your previous comments.)



##########
sdks/python/apache_beam/metrics/monitoring_infos.py:
##########
@@ -214,6 +214,11 @@ def int64_user_distribution(namespace, name, metric, ptransform=None):
     ptransform: The ptransform id used as a label.
   """
   labels = create_labels(ptransform=ptransform, namespace=namespace, name=name)
+  if metric.count <= 0:
+    raise TypeError(
+        'Expected a non zero distribution count for %s metric but received %s' %
+        (metric, metric.count))

Review Comment:
   Done.



-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2299288022

   waiting on author


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Filter empty distribution metrics [beam]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32027:
URL: https://github.com/apache/beam/pull/32027#issuecomment-2299045097

   Reminder, please take a look at this pr: @tvalentyn 


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org