Hi Maurizio, What runner are you using? Are you trying this out in the DirectRunner? As far as I know, no runner supports stateful processing in session windows yet. It is probably a bug that your pipeline was accepted by the runner when it includes features that the runner cannot execute. It would be a real mistake to have missed this validation for the DirectRunner.
Support for stateful processing in merging windows is definitely important. There's only a Jira filed for Dataflow [1] as far as I can tell. I just cloned it for the DirectRunner [2] since that is how you would test your pipeline. If you want to follow the same feature for a different runner, I can route a new ticket to the right person. Kenn [1] https://issues.apache.org/jira/browse/BEAM-2507 [2] https://issues.apache.org/jira/browse/BEAM-3686 On Fri, Feb 9, 2018 at 7:46 AM, Maurizio Sambati <[email protected]> wrote: > Hi everyone, > > I'm trying to write a simple pipeline to experiment both stateful > processing and session window. > > I have an event stream, each event has a timestamp and a session key, I > want to group by each session and enrich all events using a common state of > the session. In this case I'm just replacing the event with an incremental > counter. > > So, let's say I have a source that outputs an event every second and my > stream is [a, a, b, a, a, c, a, a, b, c, c, c, a, a] (I'm just writing only > the session key as the value is useless for the purpose of the issue I'm > experiencing) > > I want the following output: [<a, 0>, <a, 1>, <b, 0>, <a, 2>, <a, 3>, ...] > (actually the order is not important) > > Unluckily my code seems not to work as I was expecting and I'm not able to > understand the reason. (to be honest I haven't found many resources on the > topic) What I actually get is something like: > > a, 0 > a, 1 > b, 0 > a, 0 <-- ??? > a, 2, <---??? > c, 0, > ... > > that makes me wonder if I have actually understood how the state is > related to the key-window pair or maybe if I have just misunderstood how > the window/triggering works. > > My pipeline looks something like: > > p.apply(TextIO.read().from("input.json")) > > .apply(MapElements.via(new ParseTableRowJson())) > > .apply(new AugmentEvents()) > > .apply(ParDo.of(new DoFn<KV<String, Long>, Void>() { > > @ProcessElement > > public void processElement(ProcessContext c) { > > LOG.info(c.element().getKey() + ": " + c.element().getValue()); > > } > > })); > > ... > > static class AugmentEvents extends PTransform<PCollection<TableRow>, > PCollection<KV<String, Long>>> { > > @Override > > public PCollection<KV<String, Long>> expand(PCollection<TableRow> input) > { > > return input > > .apply(ParDo.of(new ExtractSessionIdAndTimestamp())) > > .apply(new ComputeSessions()); > > } > > } > > > static class ComputeSessions extends PTransform<PCollection<KV<String, > TableRow>>, PCollection<KV<String, Long>>> { > > @Override > > public PCollection<KV<String, Long>> expand(PCollection<KV<String, > TableRow>> events) { > > return events > > .apply(Window.<KV<String, TableRow>>into(Sessions.withGa > pDuration(Duration.standardMinutes(10))) > > .triggering(AfterPane.elementCountAtLeast(1)) > > .discardingFiredPanes() > > .withAllowedLateness(Duration.standardMinutes(10))) > > .apply(ParDo.of(new StatefulCount())); > > } > > } > > static class StatefulCount extends DoFn<KV<String, TableRow>, KV<String, > Long>> { > > @StateId("storage") > > private final StateSpec<ValueState<Integer>> storageSpec = > StateSpecs.value(VarIntCoder.of()); > > @ProcessElement > > public void processElement(ProcessContext context, BoundedWindow window > , @StateId("storage") ValueState<Integer> storage) { > > Integer val = storage.read(); > > if (val == null) { > > val = new Integer(0); > > } > > int current = val.intValue(); > > context.output(KV.of(context.element().getKey(), new Long(current))); > > storage.write(current+1); > > } > > } > > Maurizio > > >
