Realtime Compute for Apache Flink:Use the BITMAP type in Flink for exact deduplication

DDL

-- Data source table
CREATE TEMPORARY TABLE user_events
(
      user_id     INT
      ,tag        STRING  --  Event tag
      ,event_time TIMESTAMP(3)
      ,WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
)
WITH ('connector' = 'kafka' ...)
;

-- Sink table for minute-level UV statistics
CREATE TEMPORARY TABLE minute_uv
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING  --  Event tag
      ,minute_uv BIGINT  -- UV count
)
WITH ('connector' = 'jdbc' ...)
;

DML

• Core logic: Use the BITMAP_BUILD_CARDINALITY_AGG function to perform deduplication and counting simultaneously when the window calculation is complete.

• Scenarios: Real-time dashboards and simple minute-level monitoring metrics.

• Note: The output is a scalar value and is not additive. Because of the principles of set deduplication, you cannot directly add the statistical values from different time windows to obtain the total UV for a longer time span. For example, you cannot simply add two "1-minute UV" counts to obtain a "2-minute deduplicated UV" count.

INSERT INTO minute_uv
SELECT
    DATE_FORMAT(window_start, 'yyyy-MM-dd') AS ts_date,
    DATE_FORMAT(window_start, 'HH') AS ts_hour,
    DATE_FORMAT(window_start, 'mm') AS ts_minute,
    tag,
    -- Core function: Builds a bitmap and directly returns its cardinality (the deduplicated count).
    -- This logic is equivalent to COUNT(DISTINCT user_id) but offers better performance.
    BITMAP_BUILD_CARDINALITY_AGG(user_id) AS uv
FROM TABLE(
    TUMBLE(
        TABLE user_events, 
        DESCRIPTOR(event_time), 
        INTERVAL '1' MINUTE
    )
)
GROUP BY
    window_start,
    window_end,
    tag;

DDL

-- Data source table
CREATE TEMPORARY TABLE user_events
(
      user_id     INT
      ,tag        STRING  --  Event tag
      ,event_time TIMESTAMP(3)
      ,WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
)
WITH ('connector' = 'kafka' ...)
;

-- Sink table for minute-level bitmaps
CREATE TEMPORARY TABLE minute_bitmaps
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING  --  Event tag
      ,minute_bm BYTES   --  Serialized minute-level bitmap
)
WITH ('connector' = 'jdbc' ...)
;

DML

• Core logic: Perform a lightweight transformation from detailed data to a bitmap on the Flink side. Instead of directly outputting a count value, you can output the serialized binary data.

• Technical details: Use BITMAP_BUILD_AGG to construct a bitmap object. Then, use BITMAP_TO_BYTES to serialize it into a byte array (VARBINARY/BYTES) that complies with the RoaringBitmap standard.

• Scenarios: Building an intermediate data warehouse (DWD/DWS) or connecting to an Online Analytical Processing (OLAP) engine that supports bitmaps, such as StarRocks or Hologres.

• Key advantage: It is additive. The output binary bitmap data retains the set information. Downstream jobs or query engines can load this data at any time. They can then use a bitmap union (OR) operation to merge multiple minute-level bitmaps into hour-level or day-level bitmaps. This allows for flexible roll-ups across time windows.

INSERT INTO minute_bitmaps
SELECT
    DATE_FORMAT(window_start, 'yyyy-MM-dd') AS ts_date,
    DATE_FORMAT(window_start, 'HH') AS ts_hour,
    DATE_FORMAT(window_start, 'mm') AS ts_minute,
    tag,
    -- Core process: Construct and serialize the bitmap.
    -- 1. BITMAP_BUILD_AGG: Aggregates user_ids within the window into an in-memory bitmap object.
    -- 2. BITMAP_TO_BYTES: Encodes the bitmap object into a standard binary format for persistent storage.
    BITMAP_TO_BYTES(
        BITMAP_BUILD_AGG(user_id)
    ) AS user_id_bitmap
FROM TABLE(
    TUMBLE(
        TABLE user_events, 
        DESCRIPTOR(event_time), 
        INTERVAL '1' MINUTE
    )
)
GROUP BY
    window_start,
    window_end,
    tag;

DDL

-- Pre-calculated minute-level bitmap table
CREATE TEMPORARY TABLE minute_bitmaps
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING  -- Event tag
      ,minute_bm BYTES   -- Pre-calculated serialized minute-level bitmap (from Example 2)
)
WITH ('connector' = 'jdbc' ...)
;

-- Hour-level UV analysis table
CREATE TEMPORARY TABLE hour_combined_uv
(
      ts_date  STRING
      ,ts_hour STRING
      ,tag     STRING  -- Event tag
      ,uv      BIGINT
)
WITH ('connector' = 'jdbc' ...)
;

DML

• Core logic:

  1. Deserialize and roll up: Convert the binary data from the storage layer back to bitmap objects. Then, use a union operation to merge multiple minute-level bitmaps into an hour-level bitmap.

  2. Set algebra operations: Perform intersection, difference, and other operations on bitmaps of different dimensions in memory.

  3. Cardinality calculation: Count the number of elements in the final result set.

• Business scenarios: User persona analysis, funnel analysis, or specific group filtering.

• Key advantage: The computation logic is completely decoupled. Changes to business logic, such as adding a filter condition, only require modifying the SQL query. You do not need to change the upstream data generation pipeline. This provides a high degree of analytical flexibility.

INSERT INTO hour_combined_uv
SELECT
    ts_date,
    ts_hour,
    'A and B andnot C' AS metric_name,
    -- Step 3: Calculate the cardinality.
    -- Count the number of users in the final bitmap and output a BIGINT.
    BITMAP_CARDINALITY(
        -- Step 2: Perform set algebra operations.
        -- Logical expression: (Tag A ∩ Tag B) - Tag C
        BITMAP_ANDNOT(
            BITMAP_AND(hour_bm_a, hour_bm_b),
            hour_bm_c
        )
    ) AS uv
FROM (
    SELECT
        ts_date,
        ts_hour,
        -- Step 1: Deserialize and perform a time roll-up.
        -- Revert the stored binary data (BYTES) to bitmaps,
        -- and use BITMAP_OR_AGG to merge (union) all minute-level bitmaps within the hour.
        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm)) 
            FILTER (WHERE tag = 'A') AS hour_bm_a,
            
        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm)) 
            FILTER (WHERE tag = 'B') AS hour_bm_b,
            
        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm)) 
            FILTER (WHERE tag = 'C') AS hour_bm_c
    FROM minute_bitmaps
    WHERE tag IN ('A', 'B', 'C')
    GROUP BY
        ts_date,
        ts_hour
);

DDL

-- Pre-calculated minute-level bitmap table
CREATE TEMPORARY TABLE minute_bitmaps
(
      ts_date    STRING
      ,ts_hour   STRING
      ,ts_minute STRING
      ,tag       STRING  -- Page name
      ,minute_bm BYTES   -- Pre-calculated serialized minute-level bitmap
)
WITH ('connector' = 'jdbc' ...)
;

-- Pre-calculated day-level bitmap table
CREATE TEMPORARY TABLE daily_bitmaps
(
      ts_date   STRING
      ,tag      STRING  -- Page name
      ,daily_bm BYTES   -- Pre-calculated serialized day-level bitmap
)
WITH ('connector' = 'jdbc' ...)
;

-- Sink table for real-time retention metrics
CREATE TEMPORARY TABLE realtime_retention_metrics
(
      ts_date         STRING
      ,tag            STRING  -- Page name
      ,retained_users BIGINT  -- Number of return visits on the current day
      ,total_users    BIGINT  -- Total number of visitors from the previous day
      ,update_time    TIMESTAMP(3)
      ,PRIMARY KEY (ts_date, tag) NOT ENFORCED
)
WITH ('connector' = 'jdbc' ...)
;

DML

• Core logic:

  1. Real-time aggregation: Aggregate the minute-level bitmap stream from the current day in real time to create a cumulative bitmap for the day.

  2. Stream-batch join: Use a JOIN operation to associate the real-time bitmap of the current day with the full bitmap of the previous day from the storage layer.

  3. Intersection operation: Perform a BITMAP_AND (intersection) operation on the user group bitmaps for the two days to obtain the set of overlapping users.

• Scenarios: Real-time monitoring of user response to sales promotions, or analysis of daily active user stickiness in games.

• Key advantages:

  • Extremely fast computation: Traditional solutions need to join two large tables with hundreds of millions of rows to calculate COUNT DISTINCT. This process has enormous overhead and high latency. The BITMAP solution only needs to perform bitmap operations on a few binary objects, and the computation is completed in milliseconds.

  • Storage decoupling: Historical data, such as daily_bitmaps, can be stored in external systems like HDFS, OSS, or JDBC. This avoids long-term use of Flink state resources.

INSERT INTO realtime_retention_metrics
SELECT
    T.ts_date,
    T.tag,
    -- Metric 1: Real-time return visits on the current day (retained users).
    -- Logic: Users who appeared today ∩ Users who appeared yesterday.
    BITMAP_CARDINALITY(
        BITMAP_AND(
            T.today_bm,                 -- Real-time bitmap for today
            BITMAP_FROM_BYTES(Y.daily_bm) -- Historical bitmap from yesterday (requires deserialization)
        )
    ) AS retained_users,
    
    -- Metric 2: Baseline number of visitors from yesterday.
    -- Logic: Directly read the cardinality of yesterday's bitmap to use as the denominator for retention rate calculation.
    BITMAP_CARDINALITY(
        BITMAP_FROM_BYTES(Y.daily_bm)
    ) AS total_users,
    
    CURRENT_TIMESTAMP AS update_time
FROM (
    -- Subquery T: Real-time stream processing.
    -- Aggregates the minute-level slices of the current day into a "current-to-date" bitmap for today in real time.
    SELECT
        ts_date,
        tag,
        BITMAP_OR_AGG(BITMAP_FROM_BYTES(minute_bm)) AS today_bm
    FROM minute_bitmaps
    GROUP BY ts_date, tag
) T
-- Core join: Combines the stream (today) with the batch (yesterday).
INNER JOIN daily_bitmaps Y
    ON T.tag = Y.tag
    -- Join condition: Match the data from the day before T.
    AND CAST(T.ts_date AS DATE) = CAST(Y.ts_date AS DATE) + INTERVAL '1' DAY;