Skip to content

Improve scalability of the Watch transform #18459

Open

Description

@kennknowles

#3565 introduces the Watch transform http://s.apache.org/beam-watch-transform.

The implementation leaves several scalability-related TODOs:

  1. The state stores hashes and timestamps of outputs that have already been output and should be omitted from future polls. We could garbage-collect this state, e.g. dropping elements from "completed" and from addNewAsPending() if their timestamp is more than X behind the watermark.
  2. When a poll returns a huge number of elements, we don't necessarily have to add all of them into state.pending - instead we could add only N oldest elements and ignore others, relying on future poll rounds to provide them, in order to avoid blowing up the state. Combined with garbage collection of GrowthState.completed, this would make the transform scalable to very large poll results.

Imported from Jira BEAM-2680. Original Jira may contain additional context.
Reported by: jkff.

Activity

  1. zhengbuqian commented on Mar 27, 2023

    @zhengbuqian
    Contributor

    This has caused a dataflow customer to experience error: too many hashes/timestamps stored causing this ValueState to exceed max allowed single value size limit of 80MB(https://cloud.google.com/dataflow/quotas#limits).

  2. Abacn commented on Aug 26, 2026

    @Abacn
    Contributor

    Point 1 is addressed in upcoming Beam 2.77.0, by #39746 (Java) and #39461 (Python)

  3. goutamadwant commented on Sep 27, 2026

    @goutamadwant
    Contributor

    .take-issue

  4. goutamadwant commented on Sep 27, 2026

    @goutamadwant
    Contributor

    I verified this remains unresolved and have a tested Java fix ready on #40294

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions