如何在Apache Beam / Google Dataflow中使用ParseJsons？

Posted 2021-04-06

java新手在这里。我很难理解如何在我的Apache Beam管道中使用ParseJsons来将字符串PCollection解析为对象PCollection。

我的理解是我需要首先定义一个匹配json结构的类，然后使用ParseJsons将json字符串映射到该类的对象。

但是，ParseJsons文档看起来很神秘。我不确定如何使用Apache Beam实际执行转换。有人能给我一个如何解析行分隔的json字符串的快速而肮脏的例子吗？

这是我做过的尝试之一，但不幸的是语法不正确。

class Product {
  private String name = null;
  private String url = null;
}

p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
 .apply(new ParseJsons.of(Product))
 .apply("WriteCounts", TextIO.write().to(options.getOutput()));

答案

我想你想要：

PCollectoion<Product> = 
  p.apply("ReadLines", TextIO.read().from(options.getInputFile()))
   .apply(new ParseJsons.of(Product.class))
   .setCoder(SerializableCoder.of(MyPojo.class));