[
https://issues.apache.org/jira/browse/BEAM-5307?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16668683#comment-16668683
]
Kenneth Knowles edited comment on BEAM-5307 at 10/30/18 1:22 PM:
-----------------------------------------------------------------
Every element always has a window, so if you aren't assigning windows then they
are all in the global window. So in that case there is no window expiration
since the global window never ends. Using timers is the right idea.
StackOverflow is a better place for digging in to this, so I'll answer there.
was (Author: kenn):
Every element always has a window, so if you aren't assigning windows then they
are all in the global window. So in that case there is no window expiration
since the global window never ends. Using timers is the right idea. You'll want
to set the timer for a finite time in the future, not the end of the window.
StackOverflow is a better place for digging in to this, so I'll answer there.
> Allow injection of state in DoFn.FinishBundle
> ---------------------------------------------
>
> Key: BEAM-5307
> URL: https://issues.apache.org/jira/browse/BEAM-5307
> Project: Beam
> Issue Type: Improvement
> Components: sdk-java-core
> Reporter: Sander Ploegsma
> Assignee: Kenneth Knowles
> Priority: Major
> Fix For: Not applicable
>
>
> Example use case: a stateful {{DoFn}} that requires persisting its state to
> an external database. Instead of writing to the external database for each
> element, it is much more efficient to flush the state every once in a while,
> or when cleaning up.
> Currently, this is not possible because the {{@StateId}} injection is only
> available in processing functions and timers.
> [This|https://stackoverflow.com/questions/51789776/calculating-deltas-in-apache-beam-using-stateful-processing]
> might be a workaround, but I'm not even sure if it works.
> Instead, by allowing the use of {{@StateId}} inside a {{@FinishBundle}}
> function, we can make sure the internal state is persisted in all scenarios:
> {code:java}
> @FinishBundle
> public void flush(FinishBundleContext context, @StateId("myState")
> ValueState<StateObj> state) {
> repository.save(state.read());
> }
> {code}
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
