Ufdjrw

Setup:

read from pubsub -> window of 30s -> group by user -> combine -> write to cloud datastore

Problem:

I'm seeing DataStoreIO writer errors as objects with similar keys are present in the same transaction.

Question:

1. I want to understand how my pipeline combines results into bundles after a group by/combine operation. I would expect the bundle to be created for every window after the combine. But apparently, a bundle can contain more than 2 occurrences of the same user?




2. Can re-execution (retries) of bundles cause this behavior?




3. Is this bundling dependent of the runner?




4. Is deduplication an option? if so, how would I best approach that?

Note that I'm not looking for a replacement for the datastore writer at the end of the pipeline, I already know that we can use a different strategy. I'm merely trying to understand how the bundling happens.

There are two answers to your question. One is specific to your use case, and the other is in general about bundling / windowing in streaming.

Specific to your pipeline

I am assuming that the 'key' for Datastore is the User ID? In that case, if you have events from the same user in more than one window, your GroupByKey or Combine operations will have one separate element for every pair of user+window.

So the question is: What are you trying to insert into datastore?

• An individual user's resulting aggregate over all time? In that case, you'd need to use a Global Window.


• A user's resulting aggregate for every 30 seconds in time? Then you need to use the window as part of the key you use to insert to datastore. Does that help / make sense?

Happy to help you design your pipeline to do what you want. Chat with me in the comments or via SO chat.

The larger question about bundling of data

Bundling strategies will vary by runner. In Dataflow, you should consider the following two factors:

• Every worker is assigned a key range. Elements for the same key will be processed by the same worker.


• Windows belong to single elements; but a bundle may contain elements from multiple windows. As an example, if the data freshness metric makes a big jump*, a number of windows may be triggered - and elements of the same key in different windows would be processed in the same bundle.

*- when can Data freshness jump suddenly? A stream with a single element with a very old timestamp, and that is very slow to process may hold the watermark for a long time. Once this element is processed, the watermark may jump a lot, to the next oldest element (Check out this lecture on watermarks ; )).