在实时系统中，如交易、风险评估和物联网分析，数据流很少同步到达。为了处理这些数据，我们需要一种基于最近可用时间戳的方法来关联事件的发生。

在这里，As-Of Joins 至关重要。RisingWave 是少数原生支持实时的 As-Of Join 的流式数据库之一，而大多数数据库都缺乏这一功能。传统系统主要提供批处理 As-Of Join，虽然适合生成历史报告，但在实时决策方面却显得力不从心。

随着RisingWave对批处理和流式即时连接的支持，用户可以高效地处理低延迟事件的处理。本文探讨了为什么实时即时连接很重要，它们在RisingWave中的工作原理，以及如何有效地实现这些连接。

一个 As-Of联接 会从一个流中检索 最晚已知的事件，这些事件发生在另一个流中某个事件的 时间戳 之前或不超过该时间戳。

与需要精确时间戳匹配的传统连接不同，as-of 连接则引入了一种时间感知的方式，确保每个事件都能映射到与其最相关的前一数据点。

许多数据库仅提供批量即期连接，其中数据被分组到固定处理时段（比如，每小时或每天）。虽然这种方法对回顾性分析有效，但在需要即时洞察力的情况下却无能为力：

• 股票交易：交易者需要实时获取最新的买价和卖价来执行交易指令。
• 风险管理：基金经理依赖于实时市场状况来作出即时决策。
• 物联网分析：必须将传感器数据即时同步来检测异常并即时触发警报。

使用实时连接，RisingWave确保事件关联在数据到达时即刻发生，提供着低延迟和高准确度。

假设有这样一个情况，我们处理股票买卖，并且需要在每次交易的时候附上当时最新的市场价格。

每个交易事件都会与交易时间之前或正好在交易时间的最新市场价格相对应。

为了保持视图的持续更新，我们在RisingWave中定义了一个物化视图，它能高效维护实时的as-of联接。

    -- 创建表叫做 `trades`，它包含以下字段：交易ID为 `BIGINT` 类型，符号为 `TEXT` 类型，交易时间为 `TIMESTAMP` 类型，交易价格为 `DECIMAL(10,2)` 类型。
    CREATE TABLE trades (  
        trade_id BIGINT,  -- 交易ID
        symbol TEXT,      -- 符号
        trade_time TIMESTAMP,  -- 交易时间
        trade_price DECIMAL(10, 2)  -- 交易价格
    ) WITH (connector='kafka', topic='trades_topic');  -- 连接器为kafka，主题为trades_topic

    -- 创建表叫做 `market_prices`，它包含以下字段：符号为 `TEXT` 类型，更新时间为 `TIMESTAMP` 类型，价格为 `DECIMAL(10,2)` 类型。
    CREATE TABLE market_prices (  
        symbol TEXT,  -- 符号
        update_time TIMESTAMP,  -- 更新时间
        price DECIMAL(10, 2)  -- 价格
    ) WITH (connector='kafka', topic='market_prices_topic');  -- 连接器为kafka，主题为market_prices_topic

    -- 创建物化视图 trade_enriched 以丰富交易数据
    CREATE MATERIALIZED VIEW trade_enriched AS  
    SELECT 选择 t.trade_id, t.symbol, t.trade_time, t.trade_price,  
           m.price AS market_price, m.update_time  
    FROM 从 trades AS t  
    ASOF JOIN market_prices AS m  
    ON t.symbol = m.symbol  
    AND t.trade_time >= m.update_time;

• RisingWave 维护 market_prices 的高效缓存。
• 当交易事件发生时，它会在 trade_time 之前检索最新的价格。
• 物化视图确保查询结果既快速又持续更新。

现在，用户可以像查询普通表一样查询**trade_enriched**，始终能获得最新的实时信息。

从trade_enriched表中选择所有符号为'AAPL'的记录，按交易时间降序排列，并限制结果为前5条。

RisingWave中的实时即席连接解锁了 高速事件关联功能，使其成为金融、交易和物联网数据分析的关键工具。与批量处理方法不同的是，RisingWave的持续的更新确保数据保持最新，并使分析结果保持及时性。

通过利用物化视图（materialized views），RisingWave 优化了状态管理，实现即时查询性能且几乎无额外负担。

如果你需要实时事件增强，RisingWave 提供了一个无缝高效的解决方案。🚀

有兴趣查看测试结果或探索其他功能，比如实时的汇率查询吗？告诉我们！