You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@beam.apache.org by "ASF GitHub Bot (Jira)" <ji...@apache.org> on 2020/09/02 00:47:00 UTC
[jira] [Work logged] (BEAM-10475) GroupIntoBatches with
Runner-determined Sharding
[ https://issues.apache.org/jira/browse/BEAM-10475?focusedWorklogId=477549&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-477549 ]
ASF GitHub Bot logged work on BEAM-10475:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 02/Sep/20 00:46
Start Date: 02/Sep/20 00:46
Worklog Time Spent: 10m
Work Description: nehsyc commented on a change in pull request #12726:
URL: https://github.com/apache/beam/pull/12726#discussion_r481510949
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -150,70 +184,95 @@ public long apply(long left, long right) {
});
this.keySpec = StateSpecs.value(inputKeyCoder);
- // prefetch every 20% of batchSize elements. Do not prefetch if batchSize is too little
+ // Prefetch every 20% of batchSize elements. Do not prefetch if batchSize is too little
this.prefetchFrequency = ((batchSize / 5) <= 1) ? Long.MAX_VALUE : (batchSize / 5);
}
@ProcessElement
public void processElement(
- @TimerId(END_OF_WINDOW_ID) Timer timer,
+ @TimerId(END_OF_WINDOW_ID) Timer windowTimer,
+ @TimerId(END_OF_BUFFERING_ID) Timer bufferingTimer,
@StateId(BATCH_ID) BagState<InputT> batch,
@StateId(NUM_ELEMENTS_IN_BATCH_ID) CombiningState<Long, long[], Long> numElementsInBatch,
@StateId(KEY_ID) ValueState<K> key,
@Element KV<K, InputT> element,
BoundedWindow window,
OutputReceiver<KV<K, Iterable<InputT>>> receiver) {
- Instant windowExpires = window.maxTimestamp().plus(allowedLateness);
-
- LOG.debug(
- "*** SET TIMER *** to point in time {} for window {}",
- windowExpires.toString(),
- window.toString());
- timer.set(windowExpires);
+ Instant windowEnds = window.maxTimestamp().plus(allowedLateness);
+ LOG.debug("*** SET TIMER *** to point in time {} for window {}", windowEnds, window);
+ windowTimer.set(windowEnds);
key.write(element.getKey());
+ LOG.debug("*** BATCH *** Add element for window {} ", window);
batch.add(element.getValue());
- LOG.debug("*** BATCH *** Add element for window {} ", window.toString());
- // blind add is supported with combiningState
+ // Blind add is supported with combiningState
numElementsInBatch.add(1L);
+
Long num = numElementsInBatch.read();
+ if (num == 1 && maxBufferingDuration.isLongerThan(Duration.ZERO)) {
+ // This is the first element in batch. Start counting buffering time if a limit was set.
+ bufferingTimer.offset(maxBufferingDuration).setRelative();
+ }
if (num % prefetchFrequency == 0) {
- // prefetch data and modify batch state (readLater() modifies this)
+ // Prefetch data and modify batch state (readLater() modifies this)
batch.readLater();
}
if (num >= batchSize) {
LOG.debug("*** END OF BATCH *** for window {}", window.toString());
- flushBatch(receiver, key, batch, numElementsInBatch);
+ flushBatch(receiver, key, batch, numElementsInBatch, bufferingTimer);
}
}
+ @OnTimer(END_OF_BUFFERING_ID)
+ public void onBufferingTimer(
+ OutputReceiver<KV<K, Iterable<InputT>>> receiver,
+ @Timestamp Instant timestamp,
+ @StateId(KEY_ID) ValueState<K> key,
+ @StateId(BATCH_ID) BagState<InputT> batch,
+ @StateId(NUM_ELEMENTS_IN_BATCH_ID) CombiningState<Long, long[], Long> numElementsInBatch,
+ @TimerId(END_OF_BUFFERING_ID) Timer bufferingTimer) {
+ LOG.debug(
+ "*** END OF BUFFERING *** for timer timestamp {} with buffering duration {}",
+ timestamp,
+ maxBufferingDuration);
+ flushBatch(receiver, key, batch, numElementsInBatch, bufferingTimer);
+ }
+
@OnTimer(END_OF_WINDOW_ID)
- public void onTimerCallback(
+ public void onWindowTimer(
OutputReceiver<KV<K, Iterable<InputT>>> receiver,
@Timestamp Instant timestamp,
@StateId(KEY_ID) ValueState<K> key,
@StateId(BATCH_ID) BagState<InputT> batch,
@StateId(NUM_ELEMENTS_IN_BATCH_ID) CombiningState<Long, long[], Long> numElementsInBatch,
+ @TimerId(END_OF_BUFFERING_ID) Timer bufferingTimer,
BoundedWindow window) {
LOG.debug(
"*** END OF WINDOW *** for timer timestamp {} in windows {}",
timestamp,
window.toString());
- flushBatch(receiver, key, batch, numElementsInBatch);
+ flushBatch(receiver, key, batch, numElementsInBatch, bufferingTimer);
}
private void flushBatch(
OutputReceiver<KV<K, Iterable<InputT>>> receiver,
ValueState<K> key,
BagState<InputT> batch,
- CombiningState<Long, long[], Long> numElementsInBatch) {
+ CombiningState<Long, long[], Long> numElementsInBatch,
+ Timer bufferingTimer) {
Iterable<InputT> values = batch.read();
- // when the timer fires, batch state might be empty
+ // When the timer fires, batch state might be empty
if (!Iterables.isEmpty(values)) {
receiver.output(KV.of(key.read(), values));
}
batch.clear();
LOG.debug("*** BATCH *** clear");
numElementsInBatch.clear();
+ // We might reach here due to batch size being reached, window expiration or buffering time
+ // limit being reached. Reset the buffering timer anyway since the state is empty now. It'll
+ // be reset again if a new element arrives before the expiration time set here.
+ if (maxBufferingDuration.isLongerThan(Duration.ZERO)) {
+ bufferingTimer.offset(maxBufferingDuration).setRelative();
Review comment:
By clearing the timer, do you mean to extend the deadline like what I am doing now or set it to something else?
I feel it's a little bit cleaner to put it in FlushBatch since we always want to reset the buffering time when a batch is flushed, whatever the reason is. wdyt?
----------------------------------------------------------------
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
Issue Time Tracking
-------------------
Worklog Id: (was: 477549)
Time Spent: 1h (was: 50m)
> GroupIntoBatches with Runner-determined Sharding
> ------------------------------------------------
>
> Key: BEAM-10475
> URL: https://issues.apache.org/jira/browse/BEAM-10475
> Project: Beam
> Issue Type: Improvement
> Components: runner-dataflow
> Reporter: Siyuan Chen
> Assignee: Siyuan Chen
> Priority: P2
> Labels: GCP, performance
> Time Spent: 1h
> Remaining Estimate: 0h
>
> [https://s.apache.org/sharded-group-into-batches|https://s.apache.org/sharded-group-into-batches__]
> Improve the existing Beam transform, GroupIntoBatches, to allow runners to choose different sharding strategies depending on how the data needs to be grouped. The goal is to help with the situation where the elements to process need to be co-located to reduce the overhead that would otherwise be incurred per element, while not losing the ability to scale the parallelism. The essential idea is to build a stateful DoFn with shardable states.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)