apache/beam · error · IllegalStateException

ApproximateUnique.PerKey requires its input to use KvCoder

Error message

ApproximateUnique.PerKey requires its input to use KvCoder

What it means

ApproximateUnique.PerKey operates per key, so it must know how to encode and decode the value part of each KV to build an ApproximateUniqueCombineFn for the values. At graph construction it checks the input PCollection's Coder; if it is not a KvCoder it throws IllegalStateException because per-value encoders cannot be extracted. Beam infers KvCoder automatically for KV outputs, so this failure usually means a coder was set or transformed explicitly and lost the KvCoder type.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateUnique.java:268

    /**
     * @see ApproximateUnique#perKey(double)
     */
    public PerKey(double estimationError) {
      if (estimationError < 0.01 || estimationError > 0.5) {
        throw new IllegalArgumentException(
            "ApproximateUnique.PerKey needs an "
                + "estimation error between 1% (0.01) and 50% (0.5).");
      }

      this.sampleSize = sampleSizeFromEstimationError(estimationError);
      this.maximumEstimationError = estimationError;
    }

    @Override
    public PCollection<KV<K, Long>> expand(PCollection<KV<K, V>> input) {
      Coder<KV<K, V>> inputCoder = input.getCoder();
      if (!(inputCoder instanceof KvCoder)) {
        throw new IllegalStateException(
            "ApproximateUnique.PerKey requires its input to use KvCoder");
      }
      @SuppressWarnings("unchecked")
      final Coder<V> coder = ((KvCoder<K, V>) inputCoder).getValueCoder();

      return input.apply(Combine.perKey(new ApproximateUniqueCombineFn<>(sampleSize, coder)));
    }

    @Override
    public void populateDisplayData(DisplayData.Builder builder) {
      super.populateDisplayData(builder);
      ApproximateUnique.populateDisplayData(builder, sampleSize, maximumEstimationError);
    }
  }

  /////////////////////////////////////////////////////////////////////////////

  /**

View on GitHub (pinned to 12126d8942)

Solutions

1. Do not override the coder on the KV input; let Beam infer KvCoder from the key and value coders
2. If you must set it explicitly, use KvCoder.of(keyCoder, valueCoder)
3. Ensure the upstream transform outputs a properly typed KV with coders inferable via CoderRegistry

Example fix

// before
output.setCoder(SerializableCoder.of(KV.class));
// after
output.setCoder(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()));

if (!(input.getCoder() instanceof KvCoder)) {
  throw new IllegalStateException("upstream PCollection must use KvCoder before ApproximateUnique.perKey");
}

Coder<KV<K,V>> c = input.getCoder();
boolean isKv = c instanceof KvCoder;

try {
  result = input.apply(ApproximateUnique.perKey(0.05));
} catch (IllegalStateException e) {
  // re-apply with explicit KvCoder on upstream
  input.setCoder(KvCoder.of(keyCoder, valueCoder));
  result = input.apply(ApproximateUnique.perKey(0.05));
}

When it happens

Trigger: Applying ApproximateUnique.perKey() to a PCollection<KV<K,V>> whose coder was set via setCoder() to a non-KvCoder, or produced by a transform that erased the coder (e.g. raw PCollection from a custom source with a generic coder).

Common situations: Custom DoFns or external sources whose output coder defaults to a generic coder instead of KvCoder; manually calling setCoder on a KV collection; test pipelines that bypass coder inference.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• Could not encode the default value %s with the provided code
• cannot encode a null Integer
• cannot encode a null Long
• cannot encode a null Short
• cannot encode a null BitSet

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