[
https://issues.apache.org/jira/browse/BEAM-9856?focusedWorklogId=431903&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-431903
]
ASF GitHub Bot logged work on BEAM-9856:
----------------------------------------
Author: ASF GitHub Bot
Created on: 07/May/20 21:32
Start Date: 07/May/20 21:32
Worklog Time Spent: 10m
Work Description: jaketf commented on a change in pull request #11596:
URL: https://github.com/apache/beam/pull/11596#discussion_r421807059
##########
File path:
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/HL7v2IO.java
##########
@@ -472,24 +548,120 @@ public void initClient() throws IOException {
this.client = new HttpHealthcareApiClient();
}
+ @GetInitialRestriction
+ public OrderedTimeRange getEarliestToLatestRestriction(@Element String
hl7v2Store)
+ throws IOException {
+ from = this.client.getEarliestHL7v2SendTime(hl7v2Store, this.filter);
+ // filters are [from, to) to match logic of OffsetRangeTracker but need
latest element to be
+ // included in results set to add an extra ms to the upper bound.
+ to = this.client.getLatestHL7v2SendTime(hl7v2Store, this.filter).plus(1);
+ return new OrderedTimeRange(from, to);
+ }
+
+ @NewTracker
+ public OrderedTimeRangeTracker newTracker(@Restriction OrderedTimeRange
timeRange) {
+ return timeRange.newTracker();
+ }
+
+ @SplitRestriction
+ public void split(
+ @Restriction OrderedTimeRange timeRange,
OutputReceiver<OrderedTimeRange> out) {
+ // TODO(jaketf) How to pick optimal values for desiredNumOffsetsPerSplit
?
Review comment:
Unfortunately, in this use case dynamic splitting would be crucial
because we can't know the distribution of data in the restriction dimension
(sendTime).
If you imagine a hospital might be much busier during daytime / weekdays
than night times weekends (though never dormant due to ICU and emergency
services). "Day time" might change base on hospital location, week days are
subject to holidays, etc.
Data distribution in sendTime may be subject to significant spikes if one of
the upstream systems populating sendTime has to backfill after a maintenance
period and doesn't responsibly set this field to event time but sets all of the
sendTimes to a short range of backfill
processing time (this is sub optimal behavior of that system but sometimes a
reality).
----------------------------------------------------------------
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:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 431903)
Time Spent: 2h 40m (was: 2.5h)
> HL7v2IO.ListHL7v2Messages should be refactored to support more parallelization
> ------------------------------------------------------------------------------
>
> Key: BEAM-9856
> URL: https://issues.apache.org/jira/browse/BEAM-9856
> Project: Beam
> Issue Type: Improvement
> Components: io-java-gcp
> Reporter: Jacob Ferriero
> Assignee: Jacob Ferriero
> Priority: Minor
> Time Spent: 2h 40m
> Remaining Estimate: 0h
>
> Currently the List Messages API paginates through in a single ProcessElement
> Call.
> However we could get a restriction based on createTime using Messages.List
> filter and orderby.
>
> This is inline with the future roadmap of HL7v2 bulk export API becomes
> available that should allow splitting on (e.g. create time dimension).
> Leveraging this bulk export might be a future optimization to explore.
>
> This could take one of two forms:
> 1. dyanmically splitable via splitable DoFn (sexy, beam idiomatic: make
> optimization the runner's problem, potentially unnecessarily complex for this
> use case )
> 2. static splitting on some time partition e.g. finding the earliest
> createTime and emitting a PCollection of 1 hour partitions and paginating
> through each hour of data w/ in the time frame that the store spans, in a
> separate ProcessElement. (easy to implement but will likely have hot keys /
> stragglers based on "busy hours")
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
