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 2021/02/25 14:52:43 UTC
[GitHub] [beam] Fokko opened a new pull request #14081: [BEAM-11865] Add readMessagesWithAttributesWithCoderAndParseFn to PubSubIO
Fokko opened a new pull request #14081:
URL: https://github.com/apache/beam/pull/14081
We need to have the attributes to decide how to parse the payload.
**Please** add a meaningful description for your change here
------------------------
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 | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.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.
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)
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Fokko commented on pull request #14081: [BEAM-11865] Add readMessagesWithAttributesWithCoderAndParseFn to PubSubIO
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #14081:
URL: https://github.com/apache/beam/pull/14081#issuecomment-786612636
Thanks for looking at this @dpcollins-google @pabloem
> I don't have decision power on PubsubIO, but I'd really rather have fewer configuration options here to maintain, not more.
I agree here, but the builder is private, so I need to this is the only way to make this possible.
> Is there any reason you can't use `readMessagesWithAttributesAndMessageId()` followed by a MapElements?
I currently use `readMessagesWithAttributes()` with a `ParDo` to do this. However, I'm running into the limits of dataflow: The job graph is too large. Please try again with a smaller job graph, or split your job into two or more smaller jobs.
A `ParDo` is actually quite overkill since the parsing of the messages is rather simple, based on the header I simply parse the message. Something that would perfectly fit in the SimpleFunction, and reducing the size of the job graph.
I need the headers since it contains the message type and the version, so I know how to parse the message.
Let me know if this is enough context, and that we can get this in. Let me know if there are any further questions or concerns.
----------------------------------------------------------------
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] Fokko closed pull request #14081: [BEAM-11865] Add readMessagesWithAttributesWithCoderAndParseFn to PubSubIO
Posted by GitBox <gi...@apache.org>.
Fokko closed pull request #14081:
URL: https://github.com/apache/beam/pull/14081
----------------------------------------------------------------
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 #14081: [BEAM-11865] Add readMessagesWithAttributesWithCoderAndParseFn to PubSubIO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #14081:
URL: https://github.com/apache/beam/pull/14081#issuecomment-786279945
hi @Fokko ! thanks for your contribution - and yes, can you talk about the motivation a little more?
----------------------------------------------------------------
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] Fokko commented on pull request #14081: [BEAM-11865] Add readMessagesWithAttributesWithCoderAndParseFn to PubSubIO
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #14081:
URL: https://github.com/apache/beam/pull/14081#issuecomment-788316067
Thanks @dpcollins-google, It is a while ago when writing my last Dataflow job, and I'm getting back up to speed. I agree that that isn't the right approach for solving this problem. After your pointer, it seems like that there is a lot of schema's are being encoded in the JSON, and that this won't solve it.
Can you take a look at the following: https://github.com/apache/beam/pull/14122
----------------------------------------------------------------
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] dpcollins-google commented on pull request #14081: [BEAM-11865] Add readMessagesWithAttributesWithCoderAndParseFn to PubSubIO
Posted by GitBox <gi...@apache.org>.
dpcollins-google commented on pull request #14081:
URL: https://github.com/apache/beam/pull/14081#issuecomment-786134853
Drive by comment: I don't have decision power on PubsubIO, but I'd really rather have fewer configuration options here to maintain, not more. Is there any reason you can't use `readMessagesWithAttributesAndMessageId()` followed by a `MapElements`?
----------------------------------------------------------------
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] dpcollins-google commented on pull request #14081: [BEAM-11865] Add readMessagesWithAttributesWithCoderAndParseFn to PubSubIO
Posted by GitBox <gi...@apache.org>.
dpcollins-google commented on pull request #14081:
URL: https://github.com/apache/beam/pull/14081#issuecomment-786683119
> The job graph is too large
Pushing this into PubSubIO will not solve this issue. As you can see, the parse function is handled in exactly this way within the IO: https://github.com/apache/beam/blob/e81d12832ffc2a6ac0a87cd767f578993026a25c/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java#L841
What I'm suggesting is to do this in your user code:
```
PCollection<PubsubMessage> messages = PubsubIO.readMessagesWithAttributesAndMessageId();
PCollection<MyType> parsed = messages.apply(MapElements.into(new TypeDescriptor<MyType>() {}).via(MY_PARSER));
```
This will have the exact same effect as the above and will not require additions to PubsubIO.
Please see the guidance here https://cloud.google.com/dataflow/docs/guides/common-errors#job-graph-too-large for handling these errors. It is likely that in some way you are capturing a large amount of data in your parser (or another component), which will not be fixed by pushing the parser into PubsubIO.
----------------------------------------------------------------
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