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/04/27 18:26:51 UTC
[GitHub] [beam] jaketf opened a new pull request #11538: [BEAM-9831] Improve performance and ux for HL7v2IO
jaketf opened a new pull request #11538:
URL: https://github.com/apache/beam/pull/11538
This implements the following improvements for HL7v2IO based on initial testing with customer
1. emit early panes rather than blocking on pagination in ListHL7v2Messages
1. allow end users to use HL7v2MessageCoder in their pipelines by making constructor public
1. drop output only fields (keeping only data and labels) before calling ingestMessages in HL7v2IO.Write rather than expecting the user to do this. (e.g. in a read from one HL7v2 store write directly to another use case)
R: @pabloem
CC: @lastomato
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [x] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [x] 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.
- [x] Update `CHANGES.md` with noteworthy changes.
- [x] 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 commented on pull request #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-620860467
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] pabloem commented on pull request #11538: [BEAM-9831] Improve UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-622036898
Run Java PostCommit
----------------------------------------------------------------
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] jaketf commented on pull request #11538: [BEAM-9831] Improve UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
jaketf commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-621507148
@pabloem this should be good to retest now.
----------------------------------------------------------------
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 a change in pull request #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #11538:
URL: https://github.com/apache/beam/pull/11538#discussion_r416060393
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java
##########
@@ -475,7 +497,14 @@ public void initClient() throws IOException {
public void listMessages(ProcessContext context) throws IOException {
String hl7v2Store = context.element();
// Output all elements of all pages.
- this.client.getHL7v2MessageStream(hl7v2Store, this.filter).forEach(context::output);
+ HttpHealthcareApiClient.HL7v2MessagePages pages =
+ new HttpHealthcareApiClient.HL7v2MessagePages(client, hl7v2Store, this.filter);
+ long reqestTime = Instant.now().getMillis();
Review comment:
`requestTime`?
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java
##########
@@ -437,6 +444,20 @@ private Message fetchMessage(HealthcareApiClient client, String msgId)
.apply(Create.of(this.hl7v2Stores))
.apply(ParDo.of(new ListHL7v2MessagesFn(this.filter)))
.setCoder(new HL7v2MessageCoder())
+ // Listing takes a long time for each input element (HL7v2 store) because it has to
+ // paginate through results in a single thread / ProcessElement call in order to keep
+ // track of page token.
+ // Eagerly emit data on 1 second intervals so downstream processing can get started before
+ // all of the list results have been paginated through.
Review comment:
Unfortunately, this is not possible. If you are paginating from inside the single DoFn `processelement` call, the data coming out of it will only go downstream after the element is done being processed, so this windowing is not changing that in the execution.
This is because bundle execution is committed atomically, so the whole bundle executes before data can go downstream. You do touch on an interesting example, which is one of the reasons that we came up with SplittableDoFn.
Something you could try to do is:
```
PColll<HL7v2Message> pages = hl7v2Stores.apply(ParDo.of(new RetrieveAndOutputPagesFn()))
pages.apply(Reshuffle.viaRandomKey()).apply(ParDo.of(new FetchEachPageFn())
```
Though I don't know if you can actually do that : )
----------------------------------------------------------------
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] jaketf commented on a change in pull request #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
jaketf commented on a change in pull request #11538:
URL: https://github.com/apache/beam/pull/11538#discussion_r417543483
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java
##########
@@ -437,6 +444,20 @@ private Message fetchMessage(HealthcareApiClient client, String msgId)
.apply(Create.of(this.hl7v2Stores))
.apply(ParDo.of(new ListHL7v2MessagesFn(this.filter)))
.setCoder(new HL7v2MessageCoder())
+ // Listing takes a long time for each input element (HL7v2 store) because it has to
+ // paginate through results in a single thread / ProcessElement call in order to keep
+ // track of page token.
+ // Eagerly emit data on 1 second intervals so downstream processing can get started before
+ // all of the list results have been paginated through.
Review comment:
I've opened https://issues.apache.org/jira/browse/BEAM-9856 to explore how this could be done with splittable dofn.
----------------------------------------------------------------
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 #11538: [BEAM-9831] Improve UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-622037105
this looks good to me. I'll wait to merge after postcommit pass
----------------------------------------------------------------
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 #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-620862736
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] pabloem commented on pull request #11538: [BEAM-9831] Improve UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-622003775
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] pabloem commented on pull request #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-620162672
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] jaketf commented on a change in pull request #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
jaketf commented on a change in pull request #11538:
URL: https://github.com/apache/beam/pull/11538#discussion_r417000696
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java
##########
@@ -437,6 +444,20 @@ private Message fetchMessage(HealthcareApiClient client, String msgId)
.apply(Create.of(this.hl7v2Stores))
.apply(ParDo.of(new ListHL7v2MessagesFn(this.filter)))
.setCoder(new HL7v2MessageCoder())
+ // Listing takes a long time for each input element (HL7v2 store) because it has to
+ // paginate through results in a single thread / ProcessElement call in order to keep
+ // track of page token.
+ // Eagerly emit data on 1 second intervals so downstream processing can get started before
+ // all of the list results have been paginated through.
Review comment:
@pabloem does this mean that all of a single element's output must be buffered in memory? or will runner be smart enough to spill to disk?
Based on my initial investigation I was not able to reproduce the behavior reported by customer in a unit test.
summarized in this [gist](https://gist.github.com/jaketf/d3c2e70dde781bbb0ef1993446e34b71)
----------------------------------------------------------------
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] jaketf commented on a change in pull request #11538: [BEAM-9831] Improve UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
jaketf commented on a change in pull request #11538:
URL: https://github.com/apache/beam/pull/11538#discussion_r418142864
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java
##########
@@ -475,7 +497,14 @@ public void initClient() throws IOException {
public void listMessages(ProcessContext context) throws IOException {
String hl7v2Store = context.element();
// Output all elements of all pages.
- this.client.getHL7v2MessageStream(hl7v2Store, this.filter).forEach(context::output);
+ HttpHealthcareApiClient.HL7v2MessagePages pages =
+ new HttpHealthcareApiClient.HL7v2MessagePages(client, hl7v2Store, this.filter);
+ long reqestTime = Instant.now().getMillis();
Review comment:
This is to track latency of list pagination requests in a metric
----------------------------------------------------------------
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] jaketf commented on a change in pull request #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
jaketf commented on a change in pull request #11538:
URL: https://github.com/apache/beam/pull/11538#discussion_r416957276
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java
##########
@@ -437,6 +444,20 @@ private Message fetchMessage(HealthcareApiClient client, String msgId)
.apply(Create.of(this.hl7v2Stores))
.apply(ParDo.of(new ListHL7v2MessagesFn(this.filter)))
.setCoder(new HL7v2MessageCoder())
+ // Listing takes a long time for each input element (HL7v2 store) because it has to
+ // paginate through results in a single thread / ProcessElement call in order to keep
+ // track of page token.
+ // Eagerly emit data on 1 second intervals so downstream processing can get started before
+ // all of the list results have been paginated through.
Review comment:
Each "page" of responses is a collection of messages. It don't think it make sense to page through all the pages (dropping the real data) to then re-fetch it in the downstream parallelized step.
In testing w/ customer when pointing at an HL7v2 store with many, many messages (and therefore pages) they reported
before this change:
there was a long time before any elements were output. so long that they gave up and killed the pipeline.
after this change:
there was data coming out more regularly.
This could have been a misunderstanding or a bad test scenario.
I will try to come up with a test that reproduces this behavior.
----------------------------------------------------------------
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 #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-620859719
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] jaketf commented on pull request #11538: [BEAM-9831] Improve UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
jaketf commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-622130167
@pabloem yup. Thanks!
----------------------------------------------------------------
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 #11538: [BEAM-9831] Improve UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-622096120
ok, is this good to merge?
----------------------------------------------------------------
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] jaketf commented on pull request #11538: [BEAM-9831] Improve performance and UX for HL7v2IO
Posted by GitBox <gi...@apache.org>.
jaketf commented on pull request #11538:
URL: https://github.com/apache/beam/pull/11538#issuecomment-621494389
@pabloem Please don't merge this yet.
I'm going to remove the triggering as it's not doing what we thought it would.
I will experiment with using split-able DoFn using a createTime filter as the restriction dimension in a later PR as part of [BEAM-9856](https://issues.apache.org/jira/browse/BEAM-9856)
I'm going to add some more logic to HealthcareIOError to get http error code if relevant because this would be a useful field to filter on in queries or in advanced error handling scenarios in the DLQ
----------------------------------------------------------------
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