google-cloud-dataflow - 如何在 Dataflow 中将 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());

第 3 条语句给出了语法错误。

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>)

有人可以告诉我如何在 Dataflow 管道中将 HashMap 作为侧输入传递。

Answer 1

根据您的管道的详细信息，这里有两个不同的答案。

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

如果您有 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())