在 Apache Beam 中创建自定义窗口函数

Question

我有一个 Beam 管道，它从读取多个文本文件开始，其中文件中的每一行代表一行，稍后在管道中插入到 Bigtable 中。该场景需要确认从每个文件中提取的行数和随后插入到 Bigtable 中的行数匹配。为此，我计划开发一个自定义窗口策略，以便根据文件名将单个文件中的行分配给单个 window 作为将传递给窗口函数的键。

是否有创建自定义窗口函数的代码示例？

Answer 1

虽然我改变了确认插入行数的策略，但对于任何对窗口元素感兴趣的人来说，都是从批处理源中读取的，例如FileIO 在批处理作业中，这里是创建自定义窗口策略的代码：

public class FileWindows extends PartitioningWindowFn<Object, IntervalWindow>{

private static final long serialVersionUID = -476922142925927415L;
private static final Logger LOG = LoggerFactory.getLogger(FileWindows.class);

@Override
public IntervalWindow assignWindow(Instant timestamp) {
    Instant end = new Instant(timestamp.getMillis() + 1);
    IntervalWindow interval = new IntervalWindow(timestamp, end);
    LOG.info("FileWindows >> assignWindow(): Window assigned with Start: {}, End: {}", timestamp, end);
    return interval;
}

@Override
public boolean isCompatible(WindowFn<?, ?> other) {
    return this.equals(other);
}

@Override
public void verifyCompatibility(WindowFn<?, ?> other) throws IncompatibleWindowException {
    if (!this.isCompatible(other)) {
        throw new IncompatibleWindowException(other, String.format("Only %s objects are compatible.", FileWindows.class.getSimpleName()));
    }
  }

@Override
public Coder<IntervalWindow> windowCoder() {
    return IntervalWindow.getCoder();
}

}

然后可以在管道中使用如下：

p
 .apply("Assign_Timestamp_to_Each_Message", ParDo.of(new AssignTimestampFn()))
 .apply("Assign_Window_to_Each_Message", Window.<KV<String,String>>into(new FileWindows())
  .withAllowedLateness(Duration.standardMinutes(1))
  .discardingFiredPanes());

请记住，您需要编写 AssignTimestampFn() 以便每条消息都带有时间戳。

在 Apache Beam 中创建自定义窗口函数

Creating Custom Windowing Function in Apache Beam

dataflow

google-cloud-dataflow

apache-beam