apache/beam · error · ValueError

Deduplicate requires at least providing either…

Error message

Deduplicate requires at least providing either processing_time_duration or event_time_duration.

What it means

The Deduplicate transform deduplicates values over a time duration and throws ValueError when constructed with neither processing_time_duration nor event_time_duration. Without a duration there is no window over which to remember seen values.

Solutions

Provide a duration: pcoll | Deduplicate(processing_time_duration=timedelta(minutes=10)).
Or use event_time_duration for event-time-based dedup.
Verify any config-driven argument is not None before constructing the transform.

Example fix

// before
pcoll | Deduplicate()
// after
pcoll | Deduplicate(event_time_duration=timedelta(hours=1))

Defensive patterns

Strategy: validation

Validate before calling

def build_dedup(processing_time_duration=None, event_time_duration=None):
    if processing_time_duration is None and event_time_duration is None:
        raise ValueError('configure one duration before constructing Deduplicate')
    return Deduplicate(processing_time_duration=processing_time_duration,
                       event_time_duration=event_time_duration)

Type guard

def has_dedup_duration(**kw) -> bool:
    return kw.get('processing_time_duration') is not None or kw.get('event_time_duration') is not None

Try / catch

try:
    out = pcoll | Deduplicate(**dedup_cfg)
except ValueError as e:
    if 'Deduplicate requires' in str(e):
        out = pcoll | Deduplicate(event_time_duration=timedelta(hours=1))
    else:
        raise

Prevention

Always pass an explicit duration keyword
Assert config values are non-None before pipeline construction
Test pipeline construction (not just execution) in CI

When it happens

Trigger: Calling Deduplicate() with no arguments, or with both keyword arguments set to None, when applying it to a PCollection.

Common situations: Omitting the duration while migrating code between Deduplicate and Deduplicate.PerKey; assuming a default duration exists; passing config values that resolve to None.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/054011f851d35dc2. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/transforms/deduplicate.py:118

          self, seen_state=core.DoFn.StateParam(state_spec)):
        seen_state.clear()

    return DeduplicationFn()

  def expand(self, pcoll):
    return (
        pcoll
        | 'DeduplicateFn' >> core.ParDo(self._create_deduplicate_fn()))


class Deduplicate(ptransform.PTransform):
  """Similar to DeduplicatePerKey, the Deduplicate transform takes any arbitrary
  value as input and uses value as key to deduplicate among certain amount of
  time duration.
  """
  def __init__(self, processing_time_duration=None, event_time_duration=None):
    if processing_time_duration is None and event_time_duration is None:
      raise ValueError(
          'Deduplicate requires at least providing either '
          'processing_time_duration or event_time_duration.')
    self.processing_time_duration = processing_time_duration
    self.event_time_duration = event_time_duration

  def expand(self, pcoll):
    return (
        pcoll
        | 'Use Value as Key' >> core.Map(lambda x: (x, None))
        | 'DeduplicatePerKey' >> DeduplicatePerKey(
            processing_time_duration=self.processing_time_duration,
            event_time_duration=self.event_time_duration)
        | 'Output Value' >> core.Map(lambda kv: kv[0]))

View on GitHub (pinned to 12126d8942)