Apache Flink - 用于保持计算上下文的有状态函数

Question

当我对部分数据执行一些计算时，我有一个用例，这些计算取决于上下文（中间状态）。

例如：我有一些订单并对其进行一些计算。对按交易品种字段分组的订单执行计算。

    class Order {
        LocalDateTime ts;
        String symbol;
        ....
    }

所以我决定按符号字段键入订单并为每个组保留单独的状态：

    DataStream<Order> orders = tableEnv.toDataStream(selectStatement, Order.class);
    orders.keyBy(Order::getSymbol).flatMap(new SymbolExecutionContext()).addSink(jdbcSink);

带状态的函数：

 public class SymbolExecutionContext extends RichFlatMapFunction<Order, OrderBookRow> {

    private transient ValueState<OrderBook> orderBookState;

    @Override
    public void flatMap(Order input, Collector<OrderBookRow> out) throws Exception {
        OrderBook orderBook = this.orderBookState.value();
        if (orderBook == null) {
            orderBook = new OrderBook(input.getSymbol());
            this.orderBookState.update(orderBook);
            orderBook = this.orderBookState.value();
        }
        final List<OrderBookRow> execute = OrderBookService.execute(input, orderBook);
        for (final OrderBookRow orderBookRow : execute) {
            out.collect(orderBookRow);
        }
        this.orderBookState.update(orderBook);
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<OrderBook> descriptor = new ValueStateDescriptor<OrderBook>("orderbook", OrderBook.class, null);
        orderBookState = getRuntimeContext().getState(descriptor);
    }
}

所以我为每个独特的符号创建了新的 OrderBook。 OrderBook 表示符号上下文，其中对具有相同符号的每个订单执行一些计算。

但它似乎不起作用。如果只有一个符号，它会很好地工作。如果不止一个，它会产生无效的结果（缺失或不准确），并且结果或多或少是不可预测的。

Flink 作业以批处理方式执行。

有没有更好的方法来处理这个用例？

Answer 1

您似乎仅在状态不存在时才更新订单。这是一个有效的假设吗？进来的订单不会影响订单簿吗？

Answer 2

DataStream API 用于批量执行模式时，事件先按key排序，再按时间戳排序，这样应该可以满足您的排序要求。

由于键是顺序符号，因此将处理第一个符号的所有事件（按时间排序），然后是下一个符号，依此类推。有趣的是，您说这仅在只有一个符号时才能正常工作。这是什么版本的 Flink？您如何设置表和流执行环境，以及如何配置批处理模式？

你不应该想到

orderBook = this.orderBookState.value();

作为返回对您可以更新的对象的引用。 （对于某些 Flink 的状态后端来说，这可能是偶然的，但它不是公共接口的一部分。）相反，每次你想更新状态时都应该调用 orderBookState.update(orderBook)。