{"record":{"id":"ed257801ef3c2bf2","repo":"apache/beam","slug":"deduplicateperkey-requires-at-lease-provide-eitherprocessing","errorCode":null,"errorMessage":"DeduplicatePerKey requires at lease provide eitherprocessing_time_duration or event_time_duration.","messagePattern":"DeduplicatePerKey requires at lease provide eitherprocessing_time_duration or event_time_duration\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/deduplicate.py","lineNumber":60,"sourceCode":"@typehints.with_output_types(tuple[K, V])\nclass DeduplicatePerKey(ptransform.PTransform):\n  \"\"\" A PTransform which deduplicates <key, value> pair over a time domain and\n  threshold. Values in different windows will NOT be considered duplicates of\n  each other. Deduplication is guaranteed with respect of time domain and\n  duration.\n\n  Time durations are required so as to avoid unbounded memory and/or storage\n  requirements within a runner and care might need to be used to ensure that the\n  deduplication time limit is long enough to remove duplicates but short enough\n  to not cause performance problems within a runner. Each runner may provide an\n  optimized implementation of their choice using the deduplication time domain\n  and threshold specified.\n\n  Does not preserve any order the input PCollection might have had.\n  \"\"\"\n  def __init__(self, processing_time_duration=None, event_time_duration=None):\n    if processing_time_duration is None and event_time_duration is None:\n      raise ValueError(\n          'DeduplicatePerKey requires at lease provide either'\n          'processing_time_duration or event_time_duration.')\n    self.processing_time_duration = processing_time_duration\n    self.event_time_duration = event_time_duration\n\n  def _create_deduplicate_fn(self):\n    processing_timer_spec = userstate.TimerSpec(\n        'processing_timer', TimeDomain.REAL_TIME)\n    event_timer_spec = userstate.TimerSpec('event_timer', TimeDomain.WATERMARK)\n    state_spec = userstate.BagStateSpec('seen', BooleanCoder())\n    processing_time_duration = self.processing_time_duration\n    event_time_duration = self.event_time_duration\n\n    class DeduplicationFn(core.DoFn):\n      def process(\n          self,\n          kv,\n          ts=core.DoFn.TimestampParam,","sourceCodeStart":42,"sourceCodeEnd":78,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/deduplicate.py#L42-L78","documentation":"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.","triggerScenarios":"Calling DeduplicatePerKey() (e.g. via GroupIntoBatches-style dedup on keyed PCollections) with no arguments or with both arguments explicitly None.","commonSituations":"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.","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."],"exampleFix":"// before\npcoll | Deduplicate.PerKey()\n// after\npcoll | Deduplicate.PerKey(processing_time_duration=timedelta(minutes=5))","handlingStrategy":"validation","validationCode":"def build_dedup_per_key(processing_time_duration=None, event_time_duration=None):\n    if processing_time_duration is None and event_time_duration is None:\n        raise ValueError('configure one duration before constructing DeduplicatePerKey')\n    return Deduplicate.PerKey(processing_time_duration=processing_time_duration,\n                              event_time_duration=event_time_duration)","typeGuard":"def has_dedup_duration(**kw) -> bool:\n    return kw.get('processing_time_duration') is not None or kw.get('event_time_duration') is not None","tryCatchPattern":"try:\n    out = pcoll | Deduplicate.PerKey(**dedup_cfg)\nexcept ValueError as e:\n    if 'DeduplicatePerKey requires' in str(e):\n        out = pcoll | Deduplicate.PerKey(processing_time_duration=timedelta(minutes=10))\n    else:\n        raise","preventionTips":["Always pass an explicit duration keyword","Validate config dicts before constructing transforms","Wrap transform construction in factory functions with defaults"],"tags":["apache-beam","python","deduplicate","configuration"],"backgroundTag":"missing-required-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}