Apache Beam Go SDK：如何将 PCollection<string> 转换为 PCollection<KV<string, string>>？答案

【问题标题】：Apache Beam Go SDK: how to convert PCollection<string> to PCollection<KV<string, string>>?Apache Beam Go SDK：如何将 PCollection<string> 转换为 PCollection<KV<string, string>>？
【发布时间】：2023-02-03 10:26:04
【问题描述】：

我正在使用 Apache Beam Go SDK 并且很难获得个人收藏以正确的格式进行按键分组/组合。

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

Bob, cat
Bob, dog
Carla, cat
Carla, bunny
Doug, horse

我想用GroupByKey和组合键所以我可以像这样汇总每个人的宠物：

Bob, [cat, dog]
Carla, [cat, bunny]
Doug, [horse]

如何将 PCollection<string> 转换为 PCollection<KV<string, string>>？

他们提到了类似here 的东西，但不包括聚合字符串值的代码。

我可以使用 ParDo 获取字符串键和字符串值，如下所示，但我不知道如何转换为 GroupPerKey 输入所需的 KV<string, string> 或 CoGBK<string, string> 格式。

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<KV<string, string>> 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<string,string>
values of CoGBK<string,string> cannot bind to {Value string}

【问题讨论】：

我怀疑当您使用 groupedKV 函数时会发生错误。签名应该是ProcessElement(k string, iter func(*string) bool)（模数，你从中发出的任何东西。）beam.apache.org/documentation/programming-guide/#cogroupbykey 显示了一个带有 cogbk 的示例，但它与 gbk 相同，只有一个迭代器：查看 formatCoGBKResults 函数

标签： go type-conversion apache-beam key-value

【解决方案1】：

要映射到 KV，您可以应用 MapElements 并使用 into() 来设置 KV 类型，并在 via() 逻辑中，创建一个新的 KV.of(myKey, myValue)，例如，要获取 KV<String,String>，可以使用如下代码：

    PCollection<KV<String, String>> kvPairs = linkpages.apply(MapElements.into(
        TypeDescriptors.kvs(
            TypeDescriptors.strings(),
            TypeDescriptors.strings()))
        .via(
            linkpage -> KV.of(dataFile, linkpage)));

【讨论】：

【解决方案2】：

也许你弄错了下一个 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<KV<string, string>> 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)

【讨论】：