Dataflow // High System Lag and Slow Watermark

I am in the process of testing a data pipeline that is a hybrid between a streaming pipeline with messages that point to GCS data files. The goal of the system is to allow for a pub sub message to dictate the parameters of the deduplicating, groups, and file format (csv, tsv, pipe delimited, etc...). I have the system running successfully for smaller data sets, but seems to be struggling with larger dataset.

The reason why we have such a setup is that a majority of that datasets can be grouped together. Such as a company selling multiple products that potentially can come in messages, we would want to group these products together. We leverage side inputs on these elements to allow for processing.

The data needs to be accurate, so adding discard to late data to a trigger is not allowed. All data currently has the watermark of the PubSub message which is anticipated.

Is there something I am missing in the implementation or is my expectation of performance too high?

For example:

        message = (p 
            | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(subscription=self._subscription).with_output_types(bytes)
            | 'Convert to JSON' >> beam.Map(lambda message_json: json.loads(message_json))
            | 'Add timestamp to message' >> beam.Map(apply_datetime, key='__message_dt')
            | 'timestamps' >> beam.ParDo(AddTimestampDoFn())
            | 'Window into Fixed Intervals' >> beam.WindowInto(
                windowfn=FixedWindows(600)  # 10 minutes,
            ) 

        unique_record_keys = message | "Unique Keys" >> beam.Map(
            lambda m: m.get('unique_record_keys'),
        )  # Fetch the list of attributes/key that we can use for first round of deduplication

        record_grouping_keys = message | "Record Grouping Keys" >> beam.Map(
            lambda m: m.get('record_grouping_keys')
        )  # Fetch list of keys for group elements together as they will be stored as an object with multiple entities in a list.

        data = message | 'Read all files' >> ReadAllFromDelimitedSource(file_location_key='file_locations')  # Custom PTransform that essentially is beam.io.filebasedsource.FileBasedSource pointing to a GCS file. This file is a denormalized file and we are trying to normalize it

        key_data = data | "Keyify on Unique" >> beam.ParDo(
            GenerateKey(),
            keys_list=beam.pvalue.AsList(unique_record_keys)
        )  # Generate a key for hashing to remove duplicates rows based on columns of the source files from GCS

        grouped_data = key_data | 'Group Per Unique Key' >> beam.GroupByKey()  # Hangs up here on large dataset for hours on end with watermark not passing through window and data left unprocessed

Smaller dataset that processes successfully:

Larger dataset that hangs (also note that it looks like it fires a handful of times to this step. Original dataset is only 2 million lines from the GCS file, this is definitely adding to the issue):

I have bumped the memory and CPU as it looks like it was hanging on that, but to no avail. It experienced the same issue later on.

Update: 2022-06-05

On further inspection, the code was operating correctly. After increasing memory and performing a full inspections on the datasets, the groups were missing on a certain column. On correction, data processed as expectedly in 20 minutes.