google cloud-dataflow java管道中的条件跳过

Question

我有代码：

maxHotelEntry.apply("Convert to string", ToString.elements()).apply("Write to file", TextIO.write().to( config.getString( "gcs.checkPointLocation")).withoutSharding());

这个PCollection可以为空，-2147483648(Integer.minvalue), 124526 (+value)

如果 checkPointLocation 为空或其值小于 0，我不想写入。

一个选项是在 DoFn 中写入 GCS，但我不知道该怎么做。

Answer 1

添加一个额外的步骤来过滤掉你不想写的元素

maxHotelEntry
    .apply("Convert to string", ToString.elements())
    .apply("Filter out elements", ParDo
          .of(new DoFn<String, String>() {
              public void processElement(ProcessContext c) {
                String element = c.element();
                if ( /*perform your filter here*/ ) {
                  c.output(element);
                }
              }
          }))
    .apply("Write to file", TextIO.write().to( config.getString( "gcs.checkPointLocation")).withoutSharding());