We are using 2.13.0 From: Kenneth Knowles <k...@apache.org> Sent: Friday, May 1, 2020 4:41 PM To: user <user@beam.apache.org> Subject: Re: Notifying the closure of a Window Period
CAUTION: This email originated from outside of D&B. Please do not click links or open attachments unless you recognize the sender and know the content is safe. Ah yes... On Fri, May 1, 2020 at 8:30 AM Truebody, Kyle <truebo...@dnb.com<mailto:truebo...@dnb.com>> wrote: Yes data will be keyed by shard. This is the trigger config we used: WindowFileNamePolicy policy = new WindowFileNamePolicy(prefix,options.getDataSource()); TextIO.Write textWriter = TextIO.write() .to(policy) .withTempDirectory(tempPrefix) .withWindowedWrites() .withNumShards(options.getShardCount()); batchCollection = batchCollection.apply("Fixed Strategy",Window.<String>into( FixedWindows.of(Utilities.resolveDuration(options.getWindowDuration()))) .triggering(AfterWatermark.pastEndOfWindow()) This trigger will close the window immediately and drop data. You probably want .trigger(Repeatedly.forever(AfterWatermark.pastEndOfWindow())) or .trigger(AfterWatermark.pastEndOfWindow().withLateFirings(AfterPane.elementCountAtLeast(1))). These have the same behavior. In more recent SDKs this pipeline should be rejected to protect you from data loss caused by the trigger "closing" and dropping data. What version of SDK are you using? Kenn .withAllowedLateness(Utilities.resolveDuration(options.getWindowLateness())) .discardingFiredPanes()).apply(textWriter); From: Kenneth Knowles <k...@apache.org<mailto:k...@apache.org>> Sent: Friday, May 1, 2020 4:19 PM To: user <user@beam.apache.org<mailto:user@beam.apache.org>> Subject: Re: Notifying the closure of a Window Period CAUTION: This email originated from outside of D&B. Please do not click links or open attachments unless you recognize the sender and know the content is safe. Is the data keyed by shardNumber? To have a unique final pane for a filename prefix, you will need to include the key in the prefix. Can you also provide the triggering configuration you are working with? Kenn On Fri, May 1, 2020 at 6:47 AM Truebody, Kyle <truebo...@dnb.com<mailto:truebo...@dnb.com>> wrote: Hi Kenn, Thanks for the response… Not sure if I under this correctly : ‘affected by the fact that windows processed independently for each key’ I put a high level example below, hope it clarifies what I am trying to ask. Is there more precise way we can get informed of the final pane of a window session has been written completely. Due to nature of coordination set up for downstream consumers, the .trigger file delivery needs to be on the completion of the absolute last pane. ``` public class WindowFileNamePolicy extends FileBasedSink.FilenamePolicy { private final ResourceId prefix; private final String dataSource; /** * file names - file source name * - timestamp (processing timestamp / event timestamp) Based on the current time window * - optional : - shard number * - window start ts * @param prefix */ public WindowFileNamePolicy(ResourceId prefix,String dataSource){ this.prefix = prefix; this.dataSource = dataSource; } public String filenamePrefixForWindow(IntervalWindow window) { String filePrefix = prefix.isDirectory() ? "" : prefix.getFilename(); DateTimeFormatter formatter = DateTimeFormat.forPattern(Utilities.lngDateFormat); DateTime windowStart = formatter.parseDateTime(window.start().toString()); DateTimeFormatter resultformat = DateTimeFormat.forPattern(Utilities.shtDateFormat); return String.format( "%s/%s/%s-%s", resultformat.print(windowStart), dataSource, dataSource, resultformat.print(windowStart)); } @Override public ResourceId windowedFilename(int shardNumber, int numShards, BoundedWindow window, PaneInfo paneInfo, FileBasedSink.OutputFileHints outputFileHints) { IntervalWindow intervalWindow = (IntervalWindow) window; String filename = String.format( "%s-%s-%s", filenamePrefixForWindow(intervalWindow), shardNumber, numShards); if(paneInfo.isLast()) createTriggerFile(/*tigger file name*/ ".trigger"); //writes to the same directory of the current window. This fires multiple time depending on the number of panes that have isLast() is true/ or write operators (not sure exactly). return prefix.getCurrentDirectory().resolve(filename, ResolveOptions.StandardResolveOptions.RESOLVE_FILE); } ``` Thanks, Kyle From: Kenneth Knowles <k...@apache.org<mailto:k...@apache.org>> Sent: Friday, May 1, 2020 2:25 PM To: user <user@beam.apache.org<mailto:user@beam.apache.org>> Subject: Re: Notifying the closure of a Window Period CAUTION: This email originated from outside of D&B. Please do not click links or open attachments unless you recognize the sender and know the content is safe. I am guessing you will be affected by the fact that windows processed independently for each key. Is that what you are referring to when you mention multiple isLast() windows? Kenn On Fri, May 1, 2020 at 3:36 AM Truebody, Kyle <truebo...@dnb.com<mailto:truebo...@dnb.com>> wrote: Hi all, We are working on a streaming pipeline that we need to compatible with out legacy platform while we make the move over to Beam Streaming. Our legacy platform uses a co-ordination framework (oozie). Each step is in the coordination pipeline is active by the creation of a trigger file. I am looking for a beam construct or flag that will notify the Context/ driver of the closure of a Time window. We need to enable to create a trigger flag only when all the files have been emitted from set window period. We have tried creating the trigger flag using the PaneInfo.isLast() through a custom WindowFileNamePolicy. Noticed that a window has multiple Panes that will have isLast() as true. Thanks, Kyle