代码之家 › 专栏 › 技术社区 › Tomer Mor

与带有未绑定流的groupByKey组合

google-cloud-pubsub apache-beam google-cloud-dataflow

0

Tomer Mor · 技术社区 · 4 年前

管道是:

  1. PCollections<String> readTopic = PubSubIO.readString() ...
  2. PCollection<String> windowSession = readTopic.apply(Window.<String>into(Sessions
                                .withGapDuration(Duration.standardHours(1))));
  3.PCollection<KV<post, user> KVparsedPosts= windowSession.apply(ParDo.of(new 
     ParseEventFn());
  4. PCollection<KV<post, Iterable<user>> iterableKV= 
                        KVparsedPosts.apply(GroupByKey.create())
  5.PCollectionList<KV<post, Iterable<user>>listPosts = 
        iterableKV.apply(Combine.globally(new CombinePosts()).withoutDefaults())
  6.listPosts.apply(ParDo.of(new writePosts())

该输入是表示具有与该用户相关的帖子阵列的一个用户的消息, 帖子不是唯一的(同一个帖子可能与许多用户相关) 我的问题是步骤5组合,不确定如何组合KV<>列出这样的 writePost 将写信给D.B.一批帖子,而不仅仅是一个帖子,

实例 :假设我们有用户:A、B、C和帖子:1、2、3、4 下面是我们从PubSub中看到的消息:

{user:a, posts:[1,2,4]}
{user:b, posts:[2,3]}
{user:c, posts:[2,4]}

我想这样组合:

{post:1, user:[a]}
{post:2, user:[a,b,c]}
{post:3, user:[b]}
{post:4, user:[a,c]}

0 回复 | 直到 4 年前

1

3

Daniel Oliveira 4 年前

如果我正确理解了你的问题,那么你似乎已经想好了如何修改你的元素,以在你的例子中获得预期的结果,而你唯一剩下的问题是如何将多个帖子批处理在一起,对吗?

我将为您的示例(与您现有的代码相匹配)以及多个帖子的批处理讨论我建议的方法。

步骤1: 从自定义DoFn创建ParDo,其将用户和帖子的列表反转为<帖子,用户>。例如

{user:a, posts:[1,2,4]} -> {post:1, user:a}, {post:2, user:a}, {post:4, user:a}

你似乎已经在这条线上这么做了: 3. PCollection<KV<post, user> KVparsedPosts= windowSession.apply(ParDo.of(new ParseEventFn());

第2步: 执行GroupByKey,将post作为Key。例如

{post:2, user:a}, {post:2, user:b}, {post:2, user:c} -> {post:2, user:[a,b,c]}

你似乎在这条线上这么做: 4. PCollection<KV<post, Iterable<user>> iterableKV= KVparsedPosts.apply(GroupByKey.create())

步骤3(可选): 使用上面的代码,您可以一次向数据库写入一篇文章。如果要批处理多个帖子,则需要将多个帖子分组为批,这可以使用 GroupIntoBatches 使改变为了使用此转换,您需要为所有元素分配一个占位符键以进行批处理,这可以通过 WithKeys 例如

PCollection<KV<post, Iterable<user>> iterableKV =
        KVparsedPosts.apply(GroupByKey.create());

// Apply a key of 0 to every element.
PCollection<KV<Integer, KV<post, Iterable<user>>>> keyedPosts =
        iterableKV.apply(WithKeys.of(0));

// Group all elements into batches of 5 posts.
PCollection<KV<Integer, Iterable<KV<post, Iterable<user>>>>> batchedPosts =
        keyedPosts.apply(GroupIntoBatches.<Integer, KV<post, Iterable<user>>>ofSize(5));

一旦进行了批处理,就可以删除占位符键,并可以完成任何后续工作。