如何在数据流中传递 HashMap 作为侧面输入

Question

我正在尝试将 HashMap 作为数据流管道中的侧面输入传递。除了少数传递 String、Int 或 Long 的示例之外，我找不到任何示例。我的代码：

tagList = pipeline.apply(TextIO.Read.named("tagListTextRead").from("gs://mybucket/tag-list.json"));

PCollection<Map<String,TagObject>> tagMap = tagList
            .apply(ParDo.named("allTagsToTagMap").of(new Tags.BuildTagListMapFn()));


PCollectionView<Map<String, TagObject>> tagMapView =
            allTags.apply(View.<String, TagObject>asMap());

第三条语句出现语法错误。

The method apply(PTransform<? super PCollection<Map<String,TagObject>>,OutputT>) in the type 
     PCollection<Map<String,TagObject>> is not applicable for the arguments 
     (View.AsMap<String,TagObject>)

有人可以告诉我如何将 HashMap 作为数据流管道中的侧面输入传递。

Answer 1

这里有两个不同的答案，具体取决于您的管道的详细信息。

1. 如果您有，PCollection<KV<K, V>>那么您可以用来View.asMap()生成PCollectionView<Map<K, V>>. 无需Map自己构建。

2. 如果您有PCollection<Map<K, V>>一个元素，那么您可以使用View.asSingleton()侧面输入。

第一个可能是最自然的，你的代码最终看起来像

PCollectionView<Map<String, TagObject>> = pipeline
    .apply("tagListTextRead", TextIO.Read.from("gs://mybucket/tag-list.json"))
    .apply("tagsToKv", new Tags.TagToKvFunction())
    .apply("viewTags", View.<String, TagObject>asMap())

扩展它以显示中间值的类型：

PCollection<String> rawTags =
    pipeline.apply("tagListTextRead", TextIO.Read.from("gs://mybucket/tag-list.json"))

PCollection<KV<String, TagObject>> kvs =
    rawTags.apply("tagsToKv", new Tags.TagToKvFunction())

PCollectionView<Map<String, TagObject>> =
    kvs.apply("viewTags", View.<String, TagObject>asMap())