将Pubsub的每X条消息写入云存储

Question

我对云数据流/Apache Beam是个新手，所以概念/编程对我来说还是模糊的。

我想做的是让Dataflow监听Pubsub并以JSON的形式获取这种格式的消息：

{
  "productId": "...",
  "productName": "..."
}

并将其转换为：

{
  "productId": "...",
  "productName": "...",
  "sku": "...",
  "inventory": {
    "revenue": <some Double>,
    "stocks":  <some Integer>
  }
}

因此所需的步骤是：

>

（IngestFromPubsub）通过收听主题从Pubsub获取记录（1条Pubsub消息=1条记录）

（EnrichDataFromAPI）

a.将有效负载的JSON字符串反序列化为Java对象

b.通过调用外部API，使用SKU，我可以通过添加inventory属性来丰富每个记录的数据。

c.再次序列化记录。

（WriteToGCS）那么每个x号（可以参数化）记录，我需要在云存储中写入这些。还请考虑x=1这样一个简单的情况。(x=1，好主意吗？恐怕云存储写得太多了）

尽管我是一个Python的人，但我已经很难用Python实现这一点了，所以我需要用Java写。读Beam在Java的例子让我头疼，它太冗长了，很难遵循。我所理解的是，每个步骤都是一个。将应用到pCollection。

到目前为止，这是我微不足道的努力的结果：

public static void main(String[] args) {

    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    options.setStreaming(true);

    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply("IngestFromPubsub", PubsubIO.readStrings().fromTopic(options.getTopic()))
        // I don't really understand the next part, I just copied from official documentation and filled in some values
        .apply(Window.<String>into(FixedWindows.of(Duration.millis(5000)))
            .withAllowedLateness(Duration.millis(5000))
            .triggering(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.millis(1000)))
            .discardingFiredPanes()
        )
        .apply("EnrichDataFromAPI", ParDo.of(
            new DoFn<String, String>() {
                @ProcessElement
                public void processElement(ProcessContext c) {
                    c.element();
                    // help on this part, I heard I need to use Jackson but I don't know, for API HttpClient is sufficient
                    // ... deserialize, call API, serialize again ...
                    c.output(enrichedJSONString);
                }
            }
        ))
        .apply("WriteToGCS", 
            TextIO.write().withWindowedWrites().withNumShards(1).to(options.getOutput()))
    ;


    PipelineResult result = pipeline.run();
}

请填上缺少的部分，并给我一个关于窗口的提示（例如，什么是适当的配置等），以及我应该在哪些步骤中插入/应用它。

Answer 1

>

我认为您不需要在IngestFromPubSub和Enrich DataFromapi中进行任何窗口化。窗口化的目的是将在时间上附近的记录分组到窗口中，这样您就可以对它们进行聚合计算。但由于您不是在进行任何聚合计算，而是对独立处理每个记录感兴趣，因此不需要Windows。

由于您总是将一个输入记录转换为一个输出记录，因此您的EnrichDataFromapi应该是MapElements。这应该会使代码更容易。

在Apache Bean Java中有处理JSON的资源：Apache Beam stream processing of JSON data

您不一定需要使用jackson将JSON映射到Java对象。您可能可以直接操作JSON。您可以使用Java的原生JSON API来解析/操纵/序列化。