You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "rohdesamuel (via GitHub)" <gi...@apache.org> on 2023/06/28 03:03:01 UTC
[GitHub] [beam] rohdesamuel opened a new pull request, #27280: Implement Exception Sampling in the Python SDK
rohdesamuel opened a new pull request, #27280:
URL: https://github.com/apache/beam/pull/27280
This utilizes the ExecutionContext to set the current OutputSampler for the the input value.
------------------------
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://beam.apache.org/contribute/get-started-contributing/#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)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](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.
--
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
[GitHub] [beam] rohdesamuel commented on pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on PR #27280:
URL: https://github.com/apache/beam/pull/27280#issuecomment-1611784047
R: @udim
--
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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245550304
##########
sdks/python/apache_beam/runners/worker/operations.py:
##########
@@ -178,7 +177,10 @@ def __init__(self,
self.output_index = output_index
self.coder = coder
self.consumers = consumers
- self.element_sampler = element_sampler
+ self.exception_sampler = output_sampler
Review Comment:
TODO: rename exception_sampler to output_sampler
--
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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245560305
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1417,9 +1422,26 @@ def process(self, windowed_value):
try:
return self.do_fn_invoker.invoke_process(windowed_value)
except BaseException as exn:
+ self._maybe_sample_exception(exn, windowed_value)
self._reraise_augmented(exn)
return []
+ def _maybe_sample_exception(
+ self, exn: BaseException, windowed_value: Any) -> None:
+
+ if self.execution_context is None:
+ return
+
+ exception_sampler = self.execution_context.exception_sampler
+ if exception_sampler is None:
+ return
+
+ exception_sampler.sample_exception(
+ windowed_value,
+ exn,
Review Comment:
TODO: Get the exception with stacktrace
--
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
[GitHub] [beam] codecov[bot] commented on pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #27280:
URL: https://github.com/apache/beam/pull/27280#issuecomment-1612102656
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#27280](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (2eaee89) into [master](https://app.codecov.io/gh/apache/beam/commit/f433fbe65abb4b3e890066c4c57f41de73526c6a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (f433fbe) will **increase** coverage by `0.02%`.
> The diff coverage is `95.87%`.
```diff
@@ Coverage Diff @@
## master #27280 +/- ##
==========================================
+ Coverage 71.47% 71.50% +0.02%
==========================================
Files 858 858
Lines 104715 104806 +91
==========================================
+ Hits 74847 74939 +92
+ Misses 28320 28319 -1
Partials 1548 1548
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `80.83% <95.87%> (+0.02%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/runners/common.py](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9jb21tb24ucHk=) | `88.72% <93.33%> (+0.04%)` | :arrow_up: |
| [...ks/python/apache\_beam/runners/worker/operations.py](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvb3BlcmF0aW9ucy5weQ==) | `74.96% <94.11%> (+0.47%)` | :arrow_up: |
| [.../python/apache\_beam/runners/worker/data\_sampler.py](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvZGF0YV9zYW1wbGVyLnB5) | `98.62% <96.42%> (-0.48%)` | :arrow_down: |
| [...hon/apache\_beam/runners/worker/bundle\_processor.py](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvYnVuZGxlX3Byb2Nlc3Nvci5weQ==) | `94.18% <100.00%> (+0.02%)` | :arrow_up: |
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://app.codecov.io/gh/apache/beam/pull/27280?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `89.18% <100.00%> (-0.21%)` | :arrow_down: |
... and [7 files with indirect coverage changes](https://app.codecov.io/gh/apache/beam/pull/27280/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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
[GitHub] [beam] github-actions[bot] commented on pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27280:
URL: https://github.com/apache/beam/pull/27280#issuecomment-1611043073
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245578291
##########
sdks/python/apache_beam/runners/worker/data_sampler.py:
##########
@@ -136,6 +179,13 @@ def sample(self) -> None:
self.element_sampler.has_element = False
self._samples.append(self.element_sampler.el)
+ def sample_exception(
+ self, el: Any, exn: BaseException, transform_id: str,
+ instruction_id: str) -> None:
+ """Adds the given exception to the samples."""
+ self._exceptions.append(
Review Comment:
TODO: use the _samples_lock 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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245577985
##########
sdks/python/apache_beam/runners/worker/data_sampler.py:
##########
@@ -115,19 +130,47 @@ def remove_windowed_value(self, el: Union[WindowedValue, Any]) -> Any:
el = el.value
return el
- def flush(self, clear: bool = True) -> List[bytes]:
+ def flush(self, clear: bool = True) -> List[beam_fn_api_pb2.SampledElement]:
"""Returns all samples and optionally clears buffer if clear is True."""
with self._samples_lock:
+ # TODO(rohdesamuel): There can duplicates between the exceptions and
+ # samples. This happens when the OutputSampler samples during an
+ # exception. The fix is to create a OutputSampler per process bundle.
+ # Until then use a set to keep track of the elements.
+ seen = set(el for el, _ in self._exceptions)
if isinstance(self._coder_impl, WindowedValueCoderImpl):
- samples = [s for s in self._samples]
+ exceptions = [s for s in self._exceptions]
+ samples = [s for s in self._samples if s not in seen]
else:
- samples = [self.remove_windowed_value(s) for s in self._samples]
+ exceptions = [
+ (self.remove_windowed_value(a), b) for a, b in self._exceptions
+ ]
+ samples = [
+ self.remove_windowed_value(s) for s in self._samples
+ if s not in seen
+ ]
# Encode in the nested context b/c this ensures that the SDK can decode
# the bytes with the ToStringFn.
if clear:
self._samples.clear()
Review Comment:
TODO: clear _exceptions
--
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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245542603
##########
sdks/python/apache_beam/runners/worker/operations.py:
##########
@@ -178,7 +177,10 @@ def __init__(self,
self.output_index = output_index
self.coder = coder
self.consumers = consumers
- self.element_sampler = element_sampler
+ self.exception_sampler = output_sampler
+ self.element_sampler = (
+ output_sampler.element_sampler if output_sampler else None)
+ self.execution_context = None # type: Optional[ExecutionContext]
Review Comment:
TODO: change this type to annotation
--
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
[GitHub] [beam] github-actions[bot] commented on pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27280:
URL: https://github.com/apache/beam/pull/27280#issuecomment-1611785551
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245537151
##########
sdks/python/apache_beam/runners/worker/bundle_processor.py:
##########
@@ -1181,10 +1182,17 @@ def shutdown(self):
op.teardown()
-class ExecutionContext(object):
- def __init__(self):
- self.delayed_applications = [
- ] # type: List[Tuple[operations.DoOperation, common.SplitResultResidual]]
+@dataclass
+class ExecutionContext:
+ # Any splits to be processed later.
+ delayed_applications = [
+ ] # type: List[Tuple[operations.DoOperation, common.SplitResultResidual]]
Review Comment:
TODO: change this to a py3 type annotation
--
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
[GitHub] [beam] udim merged pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "udim (via GitHub)" <gi...@apache.org>.
udim merged PR #27280:
URL: https://github.com/apache/beam/pull/27280
--
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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245552014
##########
sdks/python/apache_beam/runners/common.py:
##########
@@ -1417,9 +1422,26 @@ def process(self, windowed_value):
try:
return self.do_fn_invoker.invoke_process(windowed_value)
except BaseException as exn:
+ self._maybe_sample_exception(exn, windowed_value)
self._reraise_augmented(exn)
return []
+ def _maybe_sample_exception(
+ self, exn: BaseException, windowed_value: Any) -> None:
Review Comment:
TODO: Change windowed_value type to WindowedValue[Any] (or something similar)
--
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
[GitHub] [beam] rohdesamuel commented on a diff in pull request #27280: Implement Exception Sampling in the Python SDK
Posted by "rohdesamuel (via GitHub)" <gi...@apache.org>.
rohdesamuel commented on code in PR #27280:
URL: https://github.com/apache/beam/pull/27280#discussion_r1245567770
##########
sdks/python/apache_beam/runners/worker/data_sampler.py:
##########
@@ -96,6 +110,7 @@ def __init__(
self._sample_timer = SampleTimer(sample_every_sec, self)
self.element_sampler = ElementSampler()
self.element_sampler.has_element = False
+ self._exceptions: Deque[Any] = collections.deque(maxlen=max_samples)
Review Comment:
Type should be Tuple[Any, ExceptionMetadata]
--
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