我正在与flink合作一个实时项目，我需要通过计算交易的先前交易功能来丰富每张卡的状态，如下所示：
对于每一张卡，我都有一个统计过去24小时内交易次数的功能。另一方面，我有两个数据源：
首先是一个数据库表，其中存储了昨天结束前的卡片交易。
第二，今天的交易流。
因此，第一步是从数据库中获取每张卡片的昨天交易记录，并将其存储在卡片状态。然后，第二步是用今天的事务更新这个状态，这些事务是在流中出现的，并计算它们在过去24小时内的事务数。我试图以流的形式读取数据库数据，并将其连接到today事务。所以，为了达到上述目标，我使用了richflatmap函数。但是，由于数据库数据本身不是流，因此输出不正确。richflatmap函数如下所示：

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流媒体程序中，实现上述丰富需求的正确设计模式是什么？我在stack overflow上发现了类似于我的问题的blow问题，但我无法解决我的问题，所以我决定寻求帮助：）
flink流媒体中使用静态数据集丰富数据流
任何帮助都将不胜感激。