如何利用历史数据集丰富Flink数据流

Question

我正在用Flink做一个实时项目，我需要用以前的交易丰富每一张卡的状态，以计算如下的交易特性：

对于每一张卡，我都有一个功能，可以统计过去24小时内的交易次数。另一方面，我有两个数据源：

transactionsHistory.connect(transactionsStream).flatMap(new         
RichCoFlatMapFunction<History, Tuple2<String, Transaction>,         
ExtractedFeatures>() {
    private ValueState<History> history;
    @Override
    public void open(Configuration config) throws Exception {
        this.history = getRuntimeContext().getState(new 
    ValueStateDescriptor<>("card history", History.class));
    }
    //historical data 
    @Override
    public void flatMap1(History history, 
    Collector<ExtractedFeatures> collector) throws Exception {
        this.history.update(history);
    }
    //new transactions from stream 
    @Override
    public void flatMap2(Tuple2<String, Transaction> 
    transactionTuple, Collector<ExtractedFeatures> collector) throws 
    Exception {
        History history = this.history.value();
        Transaction transaction = transactionTuple.f1;
        ArrayList<History> prevDayHistoryList = 
        history.prevDayTransactions;

        // This function returns transactions which are in 24 hours 
        //window of the current transaction and their count.
        Tuple2<ArrayList<History>, Integer> prevDayHistoryTuple = 
        findHistoricalDate(prevDayHistoryList,
                transaction.transactionLocalDate);
        prevDayHistoryList = prevDayHistoryTuple.f0;
        history.prevDayTransactions = prevDayHistoryList;
        this.history.update(history);
        ExtractedFeatures ef = new ExtractedFeatures();
        ef.updateFeatures(transaction, prevDayHistoryTuple.f1);
        collector.collect(ef);
    }
});

在Flink流中使用静态数据集丰富数据流

任何帮助都是非常感激的。

Answer 1

但是，由于数据库数据本身不是流，因此输出是不正确的。

当然，可以使用来自关系数据库的信息来丰富流数据。然而，最棘手的是如何保证浓缩数据在需要之前就被吸收了。通常，您可能需要缓冲要充实的流，直到充实数据被引导/摄取。例如，有时采取的一种方法是

1. 在禁用“要充实的流”的情况下运行应用程序
2. 一旦浓缩数据被完全摄取并以闪烁状态存储，就获取保存点
3. 在启用了要充实的流的情况下，从保存点重新启动应用程序

null

对这个用例的更好的支持计划在未来的版本中，BTW。