Hi Luke, I really appreciate the help. Thank you very much.
I will try the mentioned options and see which one fits to the requirement. Regards, Neha On Thu, Apr 16, 2020, 11:43 PM Luke Cwik <[email protected]> wrote: > If you have a timestamp in the record that if sorted would give you the > correct ordering then you can use a pipeline like: > ReadFromFilesWithTimestamps -> KV<Key, Data> -> > ParDo(@RequiresTimeSortedInput StatefulDoFn) -> ... > > The important parts here are that: > * Your runner must support the @RequiresTimeSortedInput[1] annotation > (very few runners support this to my knowledge since this has become > available in Apache Beam 2.20). > * ReadFromFilesWithTimestamps needs to assign the timestamp to each record > associated with its position based upon the order in which you want the > data arrive to the StatefulDoFn > * For each key, the StatefulDoFn would create a "flush" timer that occurs > at the end of the global window. For each element in @ProcessElement, it > would update state with the newly combined version, in the flush > timers @OnTimer method, you would output the value from state and clear > state. > * A shuffle/groupbykey will be inserted for you automatically before the > StatefulDoFn > > Another option is to use the SortValues[2] transform. It has several > caveats but worthwhile over implementing your own sorting algorithm so > reading that page is useful. > > A third option is if you have many files (and all the keys could be stored > in memory (possibly with a disk based backup), is to instead parallelize > processing of the files but each one file is wholly handled by a > single @ProcessElement call. So you would be responsible for reading in the > input, storing all the keys in "memory" and updating them as you see them > and finally when the file is done produce all the output you have buffered. > You could do this with the FileIO[3] transform since it produces a > PCollection of ReadableFile so you could process each individual file > separately. > > 1: > https://github.com/apache/beam/blob/c3bd4854e879da65060de8cd259865a9b34742c7/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java#L815 > 2: https://beam.apache.org/documentation/sdks/java-extensions/#sorter > 3: > https://beam.apache.org/releases/javadoc/2.19.0/index.html?org/apache/beam/sdk/io/FileIO.html > > On Thu, Apr 16, 2020 at 1:21 PM Vincent Marquez <[email protected]> > wrote: > >> I actually ran into the same issue, and would love some guidance! >> >> I had a list of avro files within folders in GCS, each folder >> representing a single day, and I needed to de-dupe events per day (by a >> key). I didn't want a GroupByKey to hold billions of events when it didn't >> matter, so I added a timestamp to each folder, then tried windowing. I >> thought perhaps the windowed event passed into the ReadAll would mean >> windows could proceed down the pipeline without having to load the entire >> batch of files into memory. >> >> I was wrong, and saw the same behavior of Neha. Is there a better way of >> doing this? I don't see a *technical* reason why the dataflow runner >> couldn't be implemented to have this behavior, but I could be mistaken. >> > > A lot of applications which only work via truly sequential processing > don't have much value added by using Apache Beam and you're likely better > off with using another framework or a simple for loop. You really want to > find a parallel processing paradigm within your data to get the benefits > that data parallel frameworks provide. > > >> >> >> *~Vincent* >> >> >> On Thu, Apr 16, 2020 at 10:13 AM Neha Sharma <[email protected]> >> wrote: >> >>> Hi Luke, >>> >>> It is the order the record appears in the source file. >>> >>> Basically each record corresponding to a key depends on the previous >>> occurrence of the same key and hence parallel processing does not seem to >>> be a good idea. >>> >>> Is there a possibility where bounded source + fixed window based on the >>> timestamp in the record can be used to somehow batch the whole data into >>> smaller chunks for processing and at the same time can maintain the >>> ordering provided a sorting based on the timestamp? >>> >>> Something like this: >>> >>> Read from Bounded Source -> >>> Fixed window to make smaller batches -> >>> Sorting based on timestamp -> >>> Processing >>> >>> >>> Regards, >>> Neha >>> >>> >>> On Thu, Apr 16, 2020, 6:57 PM Luke Cwik <[email protected]> wrote: >>> >>>> What do you mean by in sequential order, order across files, keys, ...? >>>> Is this an ordering that is based on data such as a timestamp of the >>>> record or the order in which the records appear in the source files? >>>> Do you have a lot of keys or very few? >>>> >>>> If you want to process all the data across all the files in sequential >>>> order with no parallelism then Apache Beam may not provide much value since >>>> its basis is all about parallel data processing. >>>> >>>> On Wed, Apr 15, 2020 at 10:30 PM Neha Sharma <[email protected]> >>>> wrote: >>>> >>>>> Hello, >>>>> >>>>> I have a use case where I have a bounded source and I am reading Avro >>>>> files from Google Cloud Storage. I am also using group by transform.The >>>>> amount of data is huge and I need to process the data in sequential order. >>>>> >>>>> But as Bounded source reads everything it seemed to be a good idea >>>>> fixed window on top of the global window. But it does not seem to be >>>>> working as expected. >>>>> >>>>> Can you please tell me how to handle such scenarios where a bounded >>>>> source with large dataset can be broken down into smaller chunks for >>>>> processing, using windows such that the window for a key should always be >>>>> processed in order. >>>>> >>>>> >>>>> Regards, >>>>> Neha >>>>> >>>>
