Apache Beam Go SDK：如何将 PCollection 转换为 PCollection>？

偷偷努力，悄无声息地变强，然后惊艳所有人！哈哈，小伙伴们又来学习啦~今天我将给大家介绍《Apache Beam Go SDK：如何将 PCollection 转换为 PCollection>？》，这篇文章主要会讲到等等知识点，不知道大家对其都有多少了解，下面我们就一起来看一吧！当然，非常希望大家能多多评论，给出合理的建议，我们一起学习，一起进步！

我正在使用 apache beam go sdk，但很难获取正确格式的 pcollection 以按键进行分组/组合。

在 pcollection 字符串中，每个键有多个记录，如下所示：

bob, cat
bob, dog
carla, cat
carla, bunny
doug, horse

我想使用groupbykey和combineperkey，这样我就可以像这样聚合每个人的宠物：

bob, [cat, dog]
carla, [cat, bunny]
doug, [horse]

如何将 pcollection 转换为 pcollection>？

他们在这里提到了类似的内容，但不包括聚合字符串值的代码。

我可以使用 pardo 获取字符串键和字符串值，如下所示，但我不知道如何转换为 groupperkey 输入所需的 kv 或 cogbk 格式。

pcolout := beam.pardo(s, func(line string) (string, string) {
  cleanstring := strings.trimspace(line)
  openingchar := ","
  istart := strings.index(cleanstring, openingchar)
  key := cleanstring[0:istart]
  value := cleanstring[istart+1:]
        
// how to convert to pcollection> before returning?
  return key, value
}, pcolin)

groupedkv := beam.groupbykey(s, pcolout)

它失败并出现以下错误。有什么建议吗？

panic:  inserting ParDo in scope root
        creating new DoFn in scope root
        binding fn main.main.func2
        binding params [{Value string} {Value string}] to input CoGBK
values of CoGBK cannot bind to {Value string}

正确答案

要映射到 kv，您可以应用 mapelements 并使用 into() 设置 kv 类型，并在 via() 逻辑中创建一个新的 kv.of(mykey, myvalue)，例如，获取 kv，使用类似这样的内容：

pcollection> kvpairs = linkpages.apply(mapelements.into(
        typedescriptors.kvs(
            typedescriptors.strings(),
            typedescriptors.strings()))
        .via(
            linkpage -> kv.of(datafile, linkpage)));

也许你弄错了下一个 pardo iter 类型

测试此代码

pcolIn := beam.CreateList(s, []string{"Bob, cat",
    "Bob, dog",
    "Carla, cat",
    "Carla, bunny",
    "Doug, horse",
})

pcolOut := beam.ParDo(s, func(line string) (string, string) {
    cleanString := strings.TrimSpace(line)
    openingChar := ","
    iStart := strings.Index(cleanString, openingChar)
    key := cleanString[0:iStart]
    value := cleanString[iStart+1:]

    // How to convert to PCollection> before returning?
    return key, value
}, pcolIn)

groupedKV := beam.GroupByKey(s, pcolOut)

beam.ParDo0(s, func(key string, iter func(*string) bool) {
    vals := []string{}
    val := ""
    for iter(&val) {
        vals = append(vals, strings.TrimSpace(val))
    }
    fmt.Println(key, vals)
}, groupedKV)

以上就是本文的全部内容了，是否有顺利帮助你解决问题？若是能给你带来学习上的帮助，请大家多多支持golang学习网！更多关于Golang的相关知识，也可关注golang学习网公众号。