随机访问地图端输入时，使用Go SDK for Apache Beam会引发类型错误

本篇文章给大家分享《随机访问地图端输入时，使用Go SDK for Apache Beam会引发类型错误》，覆盖了Golang的常见基础知识，其实一个语言的全部知识点一篇文章是不可能说完的，但希望通过这些问题，让读者对自己的掌握程度有一定的认识(B 数)，从而弥补自己的不足，更好的掌握它。

考虑以下简单的 apache beam go 程序：

package main

import (
    "context"
    "log"

    "github.com/apache/beam/sdks/v2/go/pkg/beam"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/register"
    _ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/direct"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/stats"
    "github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
)

func init() {
    register.function2x0(fn)
}

func main() {
    beam.init()

    p := beam.newpipeline()
    s := p.root()

    rows := beam.create(s, "1", "1", "2")
    counts := stats.count(s, rows)

    beam.pardo0(s, fn, rows, beam.sideinput{input: counts})

    if err := beamx.run(context.background(), p); err != nil {
        log.fatal(err)
    }
}

func fn(in string, lookup func(string) func(*int) bool) {
    var count int
    lookup("1")(&count)
}

使用直接运行器运行会导致以下错误：

2022/09/08 22:09:36 Executing pipeline with the direct runner.
2022/09/08 22:09:36 Pipeline:
2022/09/08 22:09:36 Nodes: {1: []uint8/bytes GLO}
{2: string/string GLO}
{3: KV/KV GLO}
{4: CoGBK/CoGBK GLO}
{5: KV/KV GLO}
Edges: 1: Impulse [] -> [Out: []uint8 -> {1: []uint8/bytes GLO}]
2: ParDo [In(Main): []uint8 <- {1: []uint8/bytes GLO}] -> [Out: T -> {2: string/string GLO}]
3: ParDo [In(Main): T <- {2: string/string GLO}] -> [Out: KV -> {3: KV/KV GLO}]
4: CoGBK [In(Main): KV <- {3: KV/KV GLO}] -> [Out: CoGBK -> {4: CoGBK/CoGBK GLO}]
5: Combine [In(Main): int <- {4: CoGBK/CoGBK GLO}] -> [Out: KV -> {5: KV/KV GLO}]
6: ParDo [In(Main): string <- {2: string/string GLO} In(MultiMap): KV <- {5: KV/KV GLO}] -> []
2022/09/08 22:09:36 Plan[plan]:
10: Impulse[0]
1: ParDo[main.fn] Out:[]
2: wait[1] Out:1
3: buffer[3]. wait:2 Out:1
4: Combine[stats.sumIntFn] Keyed:false Out:3
5: CoGBK. Out:4
6: Inject[0]. Out:5
7: ParDo[stats.keyedCountFn] Out:[6]
8: Multiplex. Out:[7 2]
9: ParDo[beam.createFn] Out:[8]
2022/09/08 22:09:36 wait[2] unblocked w/ 3 [false]
2022/09/08 22:09:36 panic: interface conversion: interface {} is string, not int
Full error:
while executing FinishBundle for Plan[plan]:
10: Impulse[0]
1: ParDo[main.fn] Out:[]
2: wait[1] Out:1
3: buffer[3]. wait:2 Out:1
4: Combine[stats.sumIntFn] Keyed:false Out:3
5: CoGBK. Out:4
6: Inject[0]. Out:5
7: ParDo[stats.keyedCountFn] Out:[6]
8: Multiplex. Out:[7 2]
9: ParDo[beam.createFn] Out:[8]
        caused by:
panic: interface conversion: interface {} is string, not int

为什么类型被推断为 string，而事实上它已成功注册为带有 kv 签名的侧面输入函数？

这里可以看到非常相似的方法：

https://github.com/apache/beam/blob/f0cd27596e3a59d6ee013ee2db232d24f46ad70a/sdks/go/examples/large_wordcount/large_wordcount.go#l345-l347

更新：这似乎是 direct 运行程序中的一个错误，因为在 dataflow 上运行它是有效的。

正确答案

不幸的是，目前直接运行程序无法正确处理多重地图输入。存在一个未解决的问题 (https://github.com/apache/beam/issues/21130) 来跟踪此问题，到目前为止，此问题从未得到支持。

今天关于《随机访问地图端输入时，使用Go SDK for Apache Beam会引发类型错误》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！