apache/beam · error · ValueError

DeduplicatePerKey requires at lease provide…

Error message

DeduplicatePerKey requires at lease provide eitherprocessing_time_duration or event_time_duration.

What it means

DeduplicatePerKey needs a time window over which to track seen keys; it throws ValueError when constructed with neither processing_time_duration nor event_time_duration. Without at least one duration there is no criterion for when a key can be seen again.

Solutions

Pass a duration: pcoll | DeduplicatePerKey(processing_time_duration=timedelta(minutes=10)).
Or use event-time dedup: DeduplicatePerKey(event_time_duration=timedelta(hours=1)).
Check for None being passed through a variable argument that should have been configured.

Example fix

// before
pcoll | Deduplicate.PerKey()
// after
pcoll | Deduplicate.PerKey(processing_time_duration=timedelta(minutes=5))

Defensive patterns

Strategy: validation

Validate before calling

def build_dedup_per_key(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 DeduplicatePerKey')
    return Deduplicate.PerKey(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.PerKey(**dedup_cfg)
except ValueError as e:
    if 'DeduplicatePerKey requires' in str(e):
        out = pcoll | Deduplicate.PerKey(processing_time_duration=timedelta(minutes=10))
    else:
        raise

Prevention

Always pass an explicit duration keyword
Validate config dicts before constructing transforms
Wrap transform construction in factory functions with defaults

When it happens

Trigger: Calling DeduplicatePerKey() (e.g. via GroupIntoBatches-style dedup on keyed PCollections) with no arguments or with both arguments explicitly None.

Common situations: Forgetting the duration parameter when copying the transform into new code; refactoring that dropped keyword args; confusion between processing-time and event-time dedup options.

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/ed257801ef3c2bf2. Report an issue: GitHub.

Appendix: source

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

@typehints.with_output_types(tuple[K, V])
class DeduplicatePerKey(ptransform.PTransform):
  """ A PTransform which deduplicates <key, value> pair over a time domain and
  threshold. Values in different windows will NOT be considered duplicates of
  each other. Deduplication is guaranteed with respect of time domain and
  duration.

  Time durations are required so as to avoid unbounded memory and/or storage
  requirements within a runner and care might need to be used to ensure that the
  deduplication time limit is long enough to remove duplicates but short enough
  to not cause performance problems within a runner. Each runner may provide an
  optimized implementation of their choice using the deduplication time domain
  and threshold specified.

  Does not preserve any order the input PCollection might have had.
  """
  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(
          'DeduplicatePerKey requires at lease provide either'
          'processing_time_duration or event_time_duration.')
    self.processing_time_duration = processing_time_duration
    self.event_time_duration = event_time_duration

  def _create_deduplicate_fn(self):
    processing_timer_spec = userstate.TimerSpec(
        'processing_timer', TimeDomain.REAL_TIME)
    event_timer_spec = userstate.TimerSpec('event_timer', TimeDomain.WATERMARK)
    state_spec = userstate.BagStateSpec('seen', BooleanCoder())
    processing_time_duration = self.processing_time_duration
    event_time_duration = self.event_time_duration

    class DeduplicationFn(core.DoFn):
      def process(
          self,
          kv,
          ts=core.DoFn.TimestampParam,

View on GitHub (pinned to 12126d8942)