在数据流运行器中运行管道时有时会出现 IllegalStateException

Question

(5d8e3f411b5a4ccb): java.lang.IllegalStateException: TimestampCombiner 将元素从 2017-09-25T13:53:08.725Z 移动到更早的时间 2017-09-25T13:53:08.718Z 用于窗口 [2017-09-25T13:53 :08.088Z..2017-09-25T13:53:08.719Z)

可能是什么原因？

WindowFn 代码很简单：

public class BQTablePartitionWindowFn extends NonMergingWindowFn<Object, IntervalWindow> {

/**
 * 
 */
private static final long serialVersionUID = 1L;

private IntervalWindow assignWindow(AssignContext context) {
    TableRow tableRow = (TableRow) context.element();
    String timestamp = tableRow.get(BQConstants.LOG_TIME).toString();
    String currentTime = DateUtil.getFormatedDate(new Date());
    DateTimeFormatter formatter = DateTimeFormat.forPattern(CommonConstants.DATE_FORMAT_YYYYMMDD_HHMMSS_SSS)
            .withZoneUTC();
    Instant start_point = Instant.parse(timestamp, formatter);
    Instant end_point = Instant.parse(currentTime, formatter);

    return new IntervalWindow(start_point, end_point);
};

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

@Override
public Collection<IntervalWindow> assignWindows(AssignContext c) throws Exception {
    return Arrays.asList(assignWindow(c));
}

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

@Override
public WindowMappingFn<IntervalWindow> getDefaultWindowMappingFn() {
    throw new IllegalArgumentException(
            "Attempted to get side input window for GlobalWindow from non-global WindowFn");
}

}

Answer 1

GroupByKey 的默认行为是输出带有时间戳的可迭代对象，该时间戳是窗口中允许的最大时间戳。对于您的窗口，即时间戳13:53:08.718Z。

该元素的时间戳13:53:08.725Z 不落在从13:53:08.088Z 到13:53:08.719Z 的窗口中。

您能否分享您的WindowFn 以及您拥有的任何调整时间戳的ParDo？

更新：感谢分享您的WindowFn。有几件事会引起您的问题。

1.指定窗口的开始时间不基于元素的时间戳。

您提取元素的一列并根据context.element().get(BQConstants.LOG_TIME) 的值分配窗口（忽略强制转换和解析）。从你的错误信息来看，这似乎不是context.timestamp()的实际值，而是元素的事件时间时间戳。

相反，您应该编写 WindowFn 以使用 context.timestamp()。您可以根据您的数据是否有界，以不同的方式确保时间戳是您想要的：

• 如果您的数据是有界的，您可以使用WithTimestamps 通过提取该字段来分配时间戳。
• 如果您的数据是无限的，则来源需要了解更多信息才能管理水印，因此配置取决于来源。例如，PubsubIO 从您可以指定的属性中读取时间戳。

2。指定窗口的结束时间以系统日期为准

几个问题：

• 结束时间向下取整，可能早于开始时间，导致窗口无效。
• 结束时间不确定。 Beam 的一般期望是，您将主要根据元素的时间戳（必须在窗口结束之前）确定性地分配窗口，其次是根据元素本身。分配像这样的不确定性窗口可能具有无法预料的缺点。一个已知问题是您的结果不可重现，如果您需要修复数据处理错误或对存档数据运行实验，这可能会很麻烦。这取决于您的用例，但您可能会考虑一些更面向未来的东西。

这里的目标是什么？您是否设置它只是为了提取动态目的地的端点？如果是这样，我建议您根据数据发生的时间而不是处理的时间对数据进行分区。