You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/05/27 21:42:51 UTC
[GitHub] [beam] rohdesamuel opened a new pull request #11838: [BEAM-9322] Modify the streaming cache to always have multiple outputs
rohdesamuel opened a new pull request #11838:
URL: https://github.com/apache/beam/pull/11838
Change-Id: I6a8eba4e323bf0fff318a56e44e512916c06266f
https://github.com/apache/beam/pull/11765 removes the ability to set the output id on TestStreams with single outputs. This PR circumvents this by always adding a dummy output to the TestStream so that it will always output a dict, so that we can control the output ids.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] 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/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem merged pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem merged pull request #11838:
URL: https://github.com/apache/beam/pull/11838
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-642985766
Run Portable_Python 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] rohdesamuel commented on a change in pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r432041629
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -315,6 +327,7 @@ def read_multiple(self, labels):
StreamingCacheSource(self._cache_dir, l,
self._is_cache_complete).read(tail=True)
for l in labels
+ if not [sub_l for sub_l in l if self.sentinel_label() in sub_l]
Review comment:
Sorry, I changed the PR and it looks like your comment is out of date. Can you PTAL?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-643057803
Run Portable_Python 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
robertwb commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-641555415
R: @robertwb
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a change in pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r437787906
##########
File path: sdks/python/apache_beam/testing/test_stream.py
##########
@@ -291,10 +291,10 @@ def expand(self, pbegin):
assert isinstance(pbegin, pvalue.PBegin)
self.pipeline = pbegin.pipeline
if not self.output_tags:
- self.output_tags = set([None])
+ self.output_tags = {None}
Review comment:
OK, in that case I'm fine with this.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-642985718
Run Python2_PVR_Flink 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #11838: [BEAM-9322] Modify the streaming cache to always have multiple outputs
Posted by GitBox <gi...@apache.org>.
robertwb commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-634964145
retest this please
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a change in pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r437702523
##########
File path: sdks/python/apache_beam/testing/test_stream.py
##########
@@ -291,10 +291,10 @@ def expand(self, pbegin):
assert isinstance(pbegin, pvalue.PBegin)
self.pipeline = pbegin.pipeline
if not self.output_tags:
- self.output_tags = set([None])
+ self.output_tags = {None}
Review comment:
If the user explicitly sets the output tags to {None}, they might be expecting a dict. (Specifically, they might get a set from elsewhere, and set the output tags from that set, and it would be awkward to have to check that set to determine how to interpret the result. So in this case I would do
```
if not self.output_tags:
return pvalue.PCollection(self.pipeline, is_bounded=False)
else:
return { ... for tag in self.output_tags}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KevinGG commented on a change in pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
KevinGG commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r431480466
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -315,6 +327,7 @@ def read_multiple(self, labels):
StreamingCacheSource(self._cache_dir, l,
self._is_cache_complete).read(tail=True)
for l in labels
+ if not [sub_l for sub_l in l if self.sentinel_label() in sub_l]
Review comment:
Or if a label `l` is a `list` of `str` and `*labels` is a `list` of `list` of `str`, then this makes sense.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-644860181
Run Python2_PVR_Flink 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-644917017
Run Python2_PVR_Flink 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-644455059
Run Portable_Python 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KevinGG commented on a change in pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
KevinGG commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r431469495
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -315,6 +327,7 @@ def read_multiple(self, labels):
StreamingCacheSource(self._cache_dir, l,
self._is_cache_complete).read(tail=True)
for l in labels
+ if not [sub_l for sub_l in l if self.sentinel_label() in sub_l]
Review comment:
This is a little hard to read.
Isn't a label `l` a `str`, so a `sub_l` is a character of that `str`?
I suppose `if not [sub_l for ...]` evaluates to `True` when the `[sub_l for ...]` is empty.
And the emptiness of `[sub_l for ...]` is based on whether the `sentinel_label` exists in the `sub_l`? This is where I get confused.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-644946194
we only had to believe in ourselves : D
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
robertwb commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-642155990
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] rohdesamuel commented on pull request #11838: [BEAM-9322] Modify the streaming cache to always have multiple outputs
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-634961730
R: @KevinGG
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] rohdesamuel commented on a change in pull request #11838: [BEAM-9322] Modify the streaming cache to always have multiple outputs
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r431471609
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -304,6 +304,18 @@ def read(self, *labels):
return iter([]), -1
return StreamingCache.Reader([header], [reader]).read(), 1
+ @staticmethod
+ def sentinel_label():
Review comment:
Yeah that can work, I like that because it keeps the same semantics. I'll go with the {None} alternative because the output_tags are always manually specified in the from_runner_api_parameter method.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a change in pull request #11838: [BEAM-9322] Modify the streaming cache to always have multiple outputs
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r431465968
##########
File path: sdks/python/apache_beam/runners/interactive/caching/streaming_cache.py
##########
@@ -304,6 +304,18 @@ def read(self, *labels):
return iter([]), -1
return StreamingCache.Reader([header], [reader]).read(), 1
+ @staticmethod
+ def sentinel_label():
Review comment:
Rather than introduce a sentinel label, how about returning a dict from expand iff output_tags was manually specified (or, alternatively, something other than `{None}`)?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11838:
URL: https://github.com/apache/beam/pull/11838#issuecomment-643471534
Run Portable_Python 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] rohdesamuel commented on a change in pull request #11838: [BEAM-9322] Modify the TestStream to output a dict when no output_tags are specified
Posted by GitBox <gi...@apache.org>.
rohdesamuel commented on a change in pull request #11838:
URL: https://github.com/apache/beam/pull/11838#discussion_r437784983
##########
File path: sdks/python/apache_beam/testing/test_stream.py
##########
@@ -291,10 +291,10 @@ def expand(self, pbegin):
assert isinstance(pbegin, pvalue.PBegin)
self.pipeline = pbegin.pipeline
if not self.output_tags:
- self.output_tags = set([None])
+ self.output_tags = {None}
Review comment:
This is a little harder to implement, mainly because the TestStream retrieves its output_tags from the keys of the PTransform payload holding it. This means that output_tags = None and output_tags = {None} look the same to the PTransform payload outputs as a map with a single key being None. When a TestStream is reconstructed, even if the original output_tags was unset, it will be constructed with output_tags = {None}.
I think the best we can do is to treat {None} and None the same way.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org