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

Question

我正在使用Apache Beam GoSDK，很难以正确的格式按键分组/组合PCollection。

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

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

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

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

如何转换PCollection

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

我可以使用ParDo来获取字符串键和字符串值，如下所示，但我不知道如何转换为KV

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}

Answer 1

要映射到KV，您可以应用MapElements并使用into（）来设置KV类型，并在via（）逻辑中，创建一个新的KV. of（myKey， myValue），例如，获取KV

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