Apache Beam多输出DoFn如何为特定标签分配类型提示？

错误的核心是DoFn1的类型提示未明确区分不同标签输出的类型，导致Beam类型检查器认为good分支的输出可能是Dict[str, Any]或TaggedOutput的联合类型，与DoFn2要求的Dict[str, Any]不匹配。

你需要通过Beam的类型提示工具明确指定DoFn1每个输出标签对应的类型，而非用Union混写返回类型。具体有两种常用方式：

方式一：使用@typehints.with_output_types装饰器 #

直接给DoFn标注各个输出标签的类型：

from apache_beam import typehints
from apache_beam.pvalue import TaggedOutput
from typing import Dict, Any, Iterable, Union

class DoFn1:
    @typehints.with_output_types({
        "good": Dict[str, Any],
        "bad": Any  # 替换为你实际的bad输出类型，比如str/int等
    })
    def process(self, row) -> Iterable[Union[Dict[str, Any], Any]]:
        if something:
            yield some_dict(...)
        else:
            yield TaggedOutput("bad", ...)

with_output_types会告诉Beam类型检查器：good标签对应Dict[str, Any]，bad标签对应你指定的类型，从而让pcoll["good"]的类型被正确识别。

方式二：通过DoFn的输出类型属性定义 #

给DoFn添加类型属性，明确每个输出的类型：

from apache_beam import typehints
from apache_beam.pvalue import TaggedOutput
from typing import Dict, Any, Iterable

class DoFn1:
    # 定义main输出（good）的类型
    DEFAULT_OUTPUT_TYPE = Dict[str, Any]
    # 定义bad输出的类型
    BAD_OUTPUT_TYPE = Any  # 替换为实际类型

    def process(self, row) -> Iterable[Union[Dict[str, Any], Any]]:
        if something:
            yield some_dict(...)
        else:
            yield TaggedOutput("bad", ...)

当使用with_outputs(main="good")时，Beam会读取DEFAULT_OUTPUT_TYPE作为good分支的类型，BAD_OUTPUT_TYPE对应bad标签的输出类型。

• 不需要传播标签输出类型提示，只要正确标注DoFn1的输出类型，Beam就能自动识别每个分支的类型。
• 不要在返回类型中包含TaggedOutput，它只是Beam用来标记输出标签的工具，实际输出值的类型才是需要标注的内容。

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