apache/beam · error

GroupIntoBatches: input PCollection must be KV-typed; got %v

Error message

GroupIntoBatches: input PCollection must be KV-typed; got %v

What it means

GroupIntoBatches requires a KV-typed PCollection because it groups elements by key. A non-KV input cannot be sharded/batched, so the transform panics with the actual input type in the message. Re-shape your data into KV pairs first.

Source

Thrown at sdks/go/pkg/beam/transforms/batch/batch.go:583

}

// GroupIntoBatches groups the values of the input PCollection<KV<K, V>>
// into batches of up to params.BatchSize elements (or
// params.BatchSizeBytes bytes) per key and emits them as
// PCollection<KV<K, []V>>.
//
// The input must be KV-typed. The key coder must be deterministic;
// non-deterministic key coders would corrupt state keying. Panics at
// pipeline build time on invalid params, non-KV input, zero limits, or
// a non-deterministic key coder.
func GroupIntoBatches(s beam.Scope, params Params, col beam.PCollection) beam.PCollection {
	s = s.Scope("batch.GroupIntoBatches")

	if err := params.validate(); err != nil {
		panic(fmt.Errorf("GroupIntoBatches: %w", err))
	}
	if !typex.IsKV(col.Type()) {
		panic(fmt.Errorf(
			"GroupIntoBatches: input PCollection must be KV-typed; got %v", col.Type()))
	}

	keyFT := col.Type().Components()[0]
	valFT := col.Type().Components()[1]

	if !beam.NewCoder(keyFT).IsDeterministic() {
		panic(fmt.Errorf(
			"GroupIntoBatches: key coder for type %v is not deterministic. "+
				"Register a deterministic custom coder with "+
				"coder.RegisterDeterministicCoder, or use a deterministic key "+
				"type (string, []byte, bool, integer, float)", keyFT.Type()))
	}

	sizerKind := sizerNone
	if params.BatchSizeBytes > 0 {
		if !isBuiltinSizeable(valFT.Type()) {
			panic(fmt.Errorf(

View on GitHub (pinned to 12126d8942)

Solutions

1. Map elements to KV pairs with beam.AddFixedKey or a ParDo producing KV<K,V>
2. Check the PCollection type with col.Type() before the call
3. Ensure the upstream emit uses KV, not a custom struct
4. Assert typex.IsKV(col.Type()) in tests

Example fix

// before
batched := batch.GroupIntoBatches(s, params, words)
// after
kvs := beam.ParDo(s, func(w string) (string, int) { return w, 1 }, words)
batched := batch.GroupIntoBatches(s, params, kvs)

if !typex.IsKV(col.Type()) {
    return fmt.Errorf("need KV input, got %v", col.Type())
}

func isKV(col beam.PCollection) bool { return typex.IsKV(col.Type()) }

When it happens

Trigger: Passing a PCollection of plain values (e.g. beam.Create of ints/strings) or a non-KV composite type directly to GroupIntoBatches, failing typex.IsKV(col.Type()).

Common situations: Forgetting a preceding beam.ParDo/Map that wraps elements into KV pairs; feeding output of a transform whose type changed after a refactor.

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

• A sink must inherit iobase.Sink, iobase.NativeSink, or be a
• empty pipeline
• create has no values
• invalid scope
• invalid scope

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