apache/beam · error

pcollection must be of KV type

Error message

pcollection must be of KV type: %v

What it means

beam.ValidateKVType asserts that a PCollection's element type is KV<A,B> and panics with "pcollection must be of KV type" if not. It exists because keyed transforms (CombinePerKey, MeanPerKey, per-key aggregations) only operate on key/value pairs. The message includes the offending PCollection's type so the mismatch is visible.

Solutions

1. Wrap elements as beam.KV before the transform: beam.KV{k, v} via ParDo emitting KV pairs.
2. Insert a Map/ParDo that converts your element type to beam.KV<K,V>.
3. Check col.Type() / print the PCollection type and confirm typex.IsKV holds.
4. If you don't need keying, use the non-keyed variants (Mean, Largest, etc.) instead.

Example fix

// before
summed := beam.CombinePerKey(s, ints) // ints is PCollection<int>
// after
pairs := beam.ParDo(s, func(i int) beam.KV { return beam.KV{"key", i} }, ints)
summed := beam.CombinePerKey(s, pairs)

col := beam.ParDo(s, func(x T) beam.KV { return beam.KV{k, x} }, input)
// ensure the collection is KV before CombinePerKey
// (in tests: if !typex.IsKV(col.Type()) { t.Fatal(...) })

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

defer func() {
    if r := recover(); r != nil {
        if s, ok := r.(string); ok && strings.Contains(s, "must be of KV type") {
            log.Fatalf("wrong transform input: %s", s)
        }
        panic(r)
    }
}()

When it happens

Trigger: Calling beam.CombinePerKey, MeanPerKey, LargestPerKey, SmallestPerKey, ApproximateWeightedQuantiles (or combinePerKey internally) on a PCollection that emits plain values instead of beam.KV pairs.

Common situations: Forgetting beam.KV() when building the input collection; applying a per-key stats transform directly to a PCollection<string> or PCollection<int>; upstream ParDo changed output type from KV to a plain value.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• pcollection must be of non-composite type
• err
• Nested FullValues must be nested as pointers.
• pubsubio.Write only accepts PCollections of
• received unknown value type: want a number:, got %T

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

Appendix: source

Thrown at sdks/go/pkg/beam/validate.go:31

// See the License for the specific language governing permissions and
// limitations under the License.

package beam

import (
	"fmt"
	"reflect"

	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/typex"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/internal/errors"
)

// ValidateKVType panics if the type of the PCollection is not KV<A,B>.
// It returns (A,B).
func ValidateKVType(col PCollection) (typex.FullType, typex.FullType) {
	t := col.Type()
	if !typex.IsKV(t) {
		panic(fmt.Sprintf("pcollection must be of KV type: %v", col))
	}
	return t.Components()[0], t.Components()[1]
}

// ValidateNonCompositeType panics if the type of the PCollection is not a
// composite type. It returns the type.
func ValidateNonCompositeType(col PCollection) typex.FullType {
	t := col.Type()
	if typex.IsComposite(t.Type()) {
		panic(fmt.Sprintf("pcollection must be of non-composite type: %v", col))
	}
	return t
}

// validate validates and processes the input collection and options. Private convenience
// function.
func validate(s Scope, col PCollection, opts []Option) ([]SideInput, map[string]reflect.Type, error) {
	if !s.IsValid() {

View on GitHub (pinned to 12126d8942)