Apache Beam Kotlin中DoFn泛型与变体问题求助

嘿，我之前也在Kotlin里折腾过Apache Beam的泛型问题，太懂这种晦涩错误的痛苦了！结合你的情况——想用Either跟踪无效结果，加了@JvmWildcard反而出更多问题——咱们一步步拆解解决：

1. 先搞对Either类型的定义 #

Beam对元素类型的序列化和泛型兼容性要求很高，首先得确保你的Either类是协变兼容且可序列化的：

import java.io.Serializable

sealed class Either<out L : Serializable, out R : Serializable> : Serializable {
    data class Left<out L : Serializable>(val value: L) : Either<L, Nothing>()
    data class Right<out R : Serializable>(val value: R) : Either<Nothing, R>()
}

这里加了Serializable约束（Beam要求元素可序列化），并且用out关键字声明协变，让Either能适配Beam的PCollection泛型体系。

2. 别乱用@JvmWildcard，试试@JvmSuppressWildcards #

Beam是Java编写的框架，Kotlin的泛型通配符处理和Java有差异：

• @JvmWildcard会让Kotlin生成带?的Java泛型，但Beam很多API（比如DoFn、PCollection）期望具体的泛型类型，而不是通配符，这就是加了注解反而报错的核心原因。
• 反而应该用@JvmSuppressWildcards，告诉Kotlin不要自动生成通配符，保持泛型的具体性。

比如写解析数据的DoFn时，可以这样：

class ParseRecordFn : DoFn<String, @JvmSuppressWildcards Either<InvalidRecord, ValidData>>() {
    @ProcessElement
    fun processElement(@Element input: String, output: OutputReceiver<Either<InvalidRecord, ValidData>>) {
        // 模拟解析逻辑：如果不符合格式输出Left，否则输出Right
        if (input.length < 5) {
            output.output(Either.Left(InvalidRecord(input, "长度不足")))
        } else {
            val parts = input.split(",")
            output.output(Either.Right(ValidData(parts[0], parts[1].toInt())))
        }
    }
}

// 定义你的无效/有效数据类（必须可序列化）
data class InvalidRecord(val raw: String, val reason: String) : Serializable
data class ValidData(val id: String, val value: Int) : Serializable

3. 构建管道时的注意事项 #

在Partition或者后续处理步骤中，要避免泛型歧义，尽量显式指定类型：

fun main() {
    val options = PipelineOptionsFactory.create()
    val pipeline = Pipeline.create(options)

    pipeline.apply(TextIO.read().from("input.csv"))
        .apply(ParDo.of(ParseRecordFn()))
        // 显式指定Partition的输入类型，避免泛型推断错误
        .apply(Partition.of(2) { elem: Either<InvalidRecord, ValidData> ->
            when (elem) {
                is Either.Left -> 0 // 无效结果分到索引0
                is Either.Right -> 1 // 有效结果分到索引1
            }
        })
        .let { partitions ->
            // 分别输出无效和有效结果
            partitions[0].apply(TextIO.write().to("output/invalid-records.txt"))
            partitions[1].apply(TextIO.write().to("output/valid-data.txt"))
        }

    pipeline.run().waitUntilFinish()
}

4. 排查原始错误的小技巧 #

如果还是报错，先把所有@JvmWildcard去掉，看原始错误信息是什么：

• 如果是“无法序列化”：检查Either和数据类是否实现了Serializable。
• 如果是“泛型类型不兼容”：确认Beam版本（建议用2.20+，对Kotlin兼容更好），或者在PCollection的泛型参数上添加@JvmSuppressWildcards。

内容的提问来源于stack exchange，提问作者Sander