Real-time marketing analytics means your funnel, attribution, A/B test, lifetime value, segment, and trigger metrics update within seconds of each impression, click, lead, and purchase, instead of after tonight's batch run. This guide builds the whole growth analytics stack on RisingWave, a PostgreSQL-compatible streaming database, as a set of materialized views over one behavioral event stream.
Why Marketing Metrics Decay in Value Every Hour They Wait
Most marketing stacks make teams wait. Events land in a warehouse, a nightly job rolls them up, and the dashboard refreshes the next morning. That cadence has real costs:
- Funnels. When a paid campaign launches, the marketing manager needs to know in minutes whether clicks are converting to leads at the expected rate. With hourly or daily rollups, a broken landing page burns a full day of spend.
- A/B tests. A 24-hour rollup hides early signals of a broken variant, blocks fast rollbacks, and forces engineers to ship on hunches.
- Attribution. Overnight attribution means today's bids are set on yesterday's channel credit. Last-touch defaults make it worse by systematically undervaluing the social, search, and email touches that built the intent.
- Lifetime value. A customer who made their first repeat purchase at 9 PM does not enter the VIP segment until the morning batch, and the retention offer goes out a day late.
- Triggers and segments. A cart abandonment email sent 5 minutes after abandonment converts at 3 to 5 times the rate of one sent an hour later. Retargeting audiences exported hourly keep serving ads to users who already bought.
The common fix is the same for every metric: compute it incrementally as events arrive. RisingWave does this with CREATE MATERIALIZED VIEW. You write the aggregation once, and the view updates on every row. The result is queryable over the PostgreSQL protocol from Metabase, Superset, Grafana, or your own API, or pushed to Kafka and application databases through sinks.
Architecture: One Behavioral Stream, Many Views
Every section below reads from the same small set of sources. Behavioral events from web and app come through Kafka. Touchpoints from your DSP and email platform come through Kafka. Orders arrive on Kafka or directly from the order database through PostgreSQL CDC. Reference data such as campaign targets, attribution weights, and suppression lists sits in RisingWave tables that you can load from your CDP through CDC.
CREATE SOURCE user_events (
event_id VARCHAR,
user_id VARCHAR,
session_id VARCHAR,
experiment_id VARCHAR, -- NULL when the user is not in a test
variant VARCHAR, -- 'A' is the control by convention
campaign_id VARCHAR,
channel VARCHAR, -- 'paid_search', 'paid_social', 'email', 'organic', 'direct'
event_type VARCHAR, -- 'exposure', 'page_view', 'product_view', 'add_to_cart',
-- 'checkout_start', 'purchase', 'lead', 'trial_signup', 'login'
product_id VARCHAR,
category VARCHAR,
value NUMERIC, -- revenue for purchases, product value for views
device_type VARCHAR,
event_time TIMESTAMPTZ
) WITH (
connector = 'kafka',
topic = 'user-behavior-events',
properties.bootstrap.server = 'kafka:9092',
scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;
CREATE SOURCE touchpoint_events (
touchpoint_id VARCHAR,
user_id VARCHAR,
channel VARCHAR,
campaign_id VARCHAR,
creative_id VARCHAR,
touchpoint_type VARCHAR, -- 'impression', 'click', 'email_open', 'email_click', 'visit'
cost_usd NUMERIC,
event_time TIMESTAMPTZ
) WITH (
connector = 'kafka',
topic = 'attribution.touchpoints',
properties.bootstrap.server = 'kafka:9092',
scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;
CREATE SOURCE orders (
order_id VARCHAR,
customer_id VARCHAR,
order_value NUMERIC,
channel VARCHAR,
status VARCHAR,
order_ts TIMESTAMPTZ
) WITH (
connector = 'kafka',
topic = 'orders',
properties.bootstrap.server = 'kafka:9092',
scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;
CREATE TABLE campaign_targets (
campaign_id VARCHAR PRIMARY KEY,
target_clicks BIGINT,
target_leads BIGINT,
target_purchases BIGINT,
target_revenue NUMERIC
);
RisingWave is open source under Apache 2.0, written in Rust, and checkpoints state to S3, so the views below survive restarts and you can rebuild any of them by replaying the Kafka log.
Funnel Stage Counts, Conversion Rates, and Drop-Off
Hourly tumbling windows give stage counts per campaign and channel. Everything else in the funnel section derives from this one view:
CREATE MATERIALIZED VIEW funnel_stage_metrics AS
SELECT
campaign_id,
channel,
window_start,
window_end,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'page_view') AS visitors,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'lead') AS leads,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'trial_signup') AS trial_signups,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'purchase') AS purchases,
SUM(value) FILTER (WHERE event_type = 'purchase') AS revenue
FROM TUMBLE(user_events, event_time, INTERVAL '1 HOUR')
GROUP BY campaign_id, channel, window_start, window_end;
CREATE MATERIALIZED VIEW funnel_conversion_rates AS
SELECT
f.*,
ROUND(f.leads::NUMERIC / NULLIF(f.visitors, 0) * 100, 2) AS visitor_to_lead_pct,
ROUND(f.purchases::NUMERIC / NULLIF(f.leads, 0) * 100, 2) AS lead_to_purchase_pct,
ROUND(f.purchases::NUMERIC / NULLIF(f.visitors, 0) * 100, 4) AS overall_cvr_pct,
ROUND(f.revenue / NULLIF(f.visitors, 0), 2) AS revenue_per_visitor,
ROUND(f.purchases::NUMERIC / NULLIF(t.target_purchases, 0) * 100, 1) AS purchase_target_pct,
f.visitors - f.leads AS visitor_to_lead_dropoff,
f.leads - f.purchases AS lead_to_purchase_dropoff
FROM funnel_stage_metrics f
LEFT JOIN campaign_targets t ON f.campaign_id = t.campaign_id;
The pacing column tells a campaign manager at 11 AM whether the day is tracking to target. The drop-off columns tell them which stage to fix first.
A/B Testing: Variant Conversion, Rolling Funnel, and Lift
Experiment events are ordinary behavioral events tagged with an experiment and a variant. Keep one row per user action rather than per session, keep revenue nullable, and keep the variant as a string rather than encoding it in column names so one view covers every experiment.
Lifetime conversion per variant uses COUNT(DISTINCT user_id) so a user who fires several exposure events counts once:
CREATE MATERIALIZED VIEW ab_variant_conversion AS
SELECT
experiment_id,
variant,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'exposure') AS exposures,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'add_to_cart') AS adds_to_cart,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'purchase') AS purchasers,
ROUND(COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'purchase')::NUMERIC
/ NULLIF(COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'exposure'), 0) * 100, 2) AS purchase_conv_pct,
COALESCE(SUM(value) FILTER (WHERE event_type = 'purchase'), 0) AS gross_revenue
FROM user_events
WHERE experiment_id IS NOT NULL
GROUP BY experiment_id, variant;
A rolling one-hour funnel that emits a fresh window every five minutes is what the product team watches on launch day. Recent windows show NULL conversion when no exposures fell into the slice, which is correct behavior, and NULLIF keeps the division safe:
CREATE MATERIALIZED VIEW ab_hourly_funnel AS
SELECT
experiment_id,
variant,
window_start,
window_end,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'exposure') AS exposures,
COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'purchase') AS purchasers,
ROUND(COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'purchase')::NUMERIC
/ NULLIF(COUNT(DISTINCT user_id) FILTER (WHERE event_type = 'exposure'), 0) * 100, 2) AS purchase_conv_pct
FROM HOP(user_events, event_time, INTERVAL '5 MINUTES', INTERVAL '1 HOUR')
WHERE experiment_id IS NOT NULL
GROUP BY experiment_id, variant, window_start, window_end;
Lift versus control is a self-join on the lifetime view:
CREATE MATERIALIZED VIEW ab_lift_vs_control AS
WITH control AS (
SELECT experiment_id, purchase_conv_pct AS ctrl_conv
FROM ab_variant_conversion
WHERE variant = 'A'
)
SELECT
v.experiment_id,
v.variant,
v.exposures,
v.purchasers,
v.purchase_conv_pct,
c.ctrl_conv,
ROUND(v.purchase_conv_pct - c.ctrl_conv, 2) AS abs_lift_pct_points,
ROUND((v.purchase_conv_pct - c.ctrl_conv) / NULLIF(c.ctrl_conv, 0) * 100, 2) AS rel_lift_pct
FROM ab_variant_conversion v
JOIN control c USING (experiment_id);
The lift view says how much better one arm appears. It does not say whether the difference is significant. Two sound patterns: let the view emit the counts and run a two-proportion z-test or a sequential test in a small Python or Go service, or materialize a guardrail rule such as "alert when variant B has at least N exposures and relative lift is below minus X percent" as a fast rollback signal. Do not claim significance from raw streaming SQL. The streaming layer's job is to keep the counts fresh.
Attribution: Last-Touch, Linear, Time-Decay, and Channel ROAS
Attribution joins each conversion to the touchpoints that preceded it within a lookback window and distributes credit. Because the join runs continuously, credit updates the moment the conversion lands.
Start with a conversion stream derived from purchases, then build the three models. Last-touch ranks touchpoints by recency within a 30-day window:
CREATE MATERIALIZED VIEW conversions AS
SELECT order_id AS conversion_id, customer_id AS user_id, order_value AS revenue_usd, order_ts AS converted_at
FROM orders
WHERE status = 'completed';
CREATE MATERIALIZED VIEW last_touch_attribution AS
SELECT conversion_id, user_id, revenue_usd AS attributed_revenue, channel, campaign_id, converted_at, touchpoint_time
FROM (
SELECT
cv.conversion_id, cv.user_id, cv.revenue_usd, cv.converted_at,
t.channel, t.campaign_id, t.event_time AS touchpoint_time,
ROW_NUMBER() OVER (PARTITION BY cv.conversion_id ORDER BY t.event_time DESC) AS touch_rank
FROM conversions cv
JOIN touchpoint_events t
ON cv.user_id = t.user_id
AND t.event_time BETWEEN cv.converted_at - INTERVAL '30 DAYS' AND cv.converted_at
) ranked
WHERE touch_rank = 1;
Linear attribution splits credit equally, and time-decay weights each touch by a seven-day half-life so closing channels earn more while awareness channels still get paid:
CREATE MATERIALIZED VIEW time_decay_attribution AS
WITH scored AS (
SELECT
cv.conversion_id, cv.user_id, cv.revenue_usd, cv.converted_at,
t.touchpoint_id, t.channel, t.campaign_id, t.cost_usd, t.event_time AS touchpoint_time,
POWER(0.5, EXTRACT(EPOCH FROM (cv.converted_at - t.event_time)) / (7 * 86400.0)) AS decay_weight
FROM conversions cv
JOIN touchpoint_events t
ON cv.user_id = t.user_id
AND t.event_time BETWEEN cv.converted_at - INTERVAL '30 DAYS' AND cv.converted_at
),
weighted AS (
SELECT *,
SUM(decay_weight) OVER (PARTITION BY conversion_id) AS total_weight,
COUNT(*) OVER (PARTITION BY conversion_id) AS total_touches
FROM scored
)
SELECT
conversion_id, user_id, channel, campaign_id, converted_at, touchpoint_time, cost_usd,
revenue_usd / total_touches AS linear_attributed_revenue,
decay_weight / total_weight AS decay_credit_fraction,
revenue_usd * (decay_weight / total_weight) AS decay_attributed_revenue
FROM weighted;
A daily rollup by channel gives attributed conversions, attributed revenue, spend, and attributed return on ad spend, which is the number a budget optimizer or bid manager should consume:
CREATE MATERIALIZED VIEW channel_attribution_daily AS
SELECT
channel,
campaign_id,
window_start,
window_end,
COUNT(DISTINCT conversion_id) AS attributed_conversions,
SUM(decay_attributed_revenue) AS attributed_revenue,
SUM(cost_usd) AS channel_spend,
ROUND(SUM(decay_attributed_revenue) / NULLIF(SUM(cost_usd), 0), 2) AS attributed_roas
FROM TUMBLE(time_decay_attribution, converted_at, INTERVAL '1 DAY')
GROUP BY channel, campaign_id, window_start, window_end;
Switching models is a change to one expression. Keeping two models side by side in the same view, as above, lets marketing compare how much credit moves between channels when the model changes.
Customer Journeys: Stage, Duration, and Channel Paths
Journey analytics needs the order of touches, not just their count. A seven-day tumbling window per user gives stage activity, and a derived view labels where the user is and how long they have been on the path:
CREATE MATERIALIZED VIEW journey_stage_activity AS
SELECT
user_id,
window_start,
window_end,
COUNT(*) FILTER (WHERE event_type = 'product_view') AS product_views,
COUNT(*) FILTER (WHERE event_type = 'add_to_cart') AS cart_events,
COUNT(*) FILTER (WHERE event_type = 'checkout_start') AS checkout_starts,
COUNT(*) FILTER (WHERE event_type = 'purchase') AS purchases,
COUNT(DISTINCT channel) AS channel_count,
MIN(event_time) AS first_touch,
MAX(event_time) AS last_touch
FROM TUMBLE(user_events, event_time, INTERVAL '7 DAYS')
GROUP BY user_id, window_start, window_end;
CREATE MATERIALIZED VIEW journey_stage AS
SELECT
*,
EXTRACT(EPOCH FROM (last_touch - first_touch)) / 3600 AS journey_hours,
CASE
WHEN purchases > 0 THEN 'converted'
WHEN checkout_starts > 0 THEN 'checkout_abandoned'
WHEN cart_events > 0 THEN 'cart_abandoned'
WHEN product_views > 0 THEN 'browsing'
ELSE 'awareness'
END AS stage
FROM journey_stage_activity;
Session windows keyed by user and grouped by channel show the channel sequence inside each visit, which is the raw material for path analysis:
CREATE MATERIALIZED VIEW channel_touch_sessions AS
SELECT
user_id,
channel,
window_start,
window_end,
COUNT(*) FILTER (WHERE event_type = 'page_view') AS page_views,
MAX(event_time) AS last_event
FROM SESSION(user_events, event_time, INTERVAL '30 MINUTES')
GROUP BY user_id, channel, window_start, window_end;
Customer Lifetime Value and RFM Segments
CLV components, purchase frequency, average order value, tenure, and churn likelihood, are all derivable from the order stream. Processing that stream continuously turns CLV into a live metric. SHOPLINE, a commerce platform serving merchants across Asia, moved its merchant analytics to RisingWave for exactly this reason and cut API response times by 76.7% by precomputing these values instead of aggregating at query time.
CREATE MATERIALIZED VIEW customer_purchase_metrics AS
SELECT
customer_id,
COUNT(order_id) AS purchase_count,
SUM(order_value) AS total_spend,
AVG(order_value) AS avg_order_value,
MIN(order_ts) AS first_purchase_ts,
MAX(order_ts) AS last_purchase_ts,
EXTRACT(EPOCH FROM (MAX(order_ts) - MIN(order_ts))) / 86400.0 AS tenure_days,
CASE WHEN COUNT(order_id) > 1
THEN EXTRACT(EPOCH FROM (MAX(order_ts) - MIN(order_ts))) / 86400.0 / (COUNT(order_id) - 1)
END AS avg_days_between_orders
FROM orders
WHERE status = 'completed'
GROUP BY customer_id;
CREATE MATERIALIZED VIEW customer_clv AS
SELECT
customer_id,
total_spend,
purchase_count,
avg_order_value,
tenure_days,
avg_days_between_orders,
avg_order_value * (365.0 / NULLIF(avg_days_between_orders, 0)) AS projected_annual_value,
avg_order_value * (365.0 / NULLIF(avg_days_between_orders, 0)) * 3 * 0.65 AS clv_3yr,
EXTRACT(EPOCH FROM (NOW() - last_purchase_ts)) / 86400.0
/ NULLIF(avg_days_between_orders, 0) AS churn_risk_ratio,
CASE
WHEN total_spend > 500 AND purchase_count >= 5 THEN 'platinum'
WHEN total_spend > 200 AND purchase_count >= 3 THEN 'gold'
WHEN purchase_count >= 2 THEN 'silver'
ELSE 'new'
END AS ltv_segment
FROM customer_purchase_metrics;
The 0.65 retention factor and the tier thresholds are placeholders. Replace them with your cohort retention curve and your own revenue bands. A churn_risk_ratio above 2 means the customer is twice as overdue as their usual cadence and is the natural trigger for a win-back offer.
Active Users: DAU, MAU, and Stickiness
Daily and monthly active users are the heartbeat metrics for any app, game, or subscription product. A login bug shipped at 2 PM should show up as a DAU drop by 2:15, not tomorrow. A daily tumbling window and a 30-day hopping window give both numbers, and their ratio is stickiness:
CREATE MATERIALIZED VIEW dau AS
SELECT
window_start::DATE AS activity_date,
COUNT(DISTINCT user_id) AS dau,
COUNT(DISTINCT user_id) FILTER (WHERE device_type = 'mobile') AS dau_mobile,
COUNT(DISTINCT user_id) FILTER (WHERE device_type = 'desktop') AS dau_desktop
FROM TUMBLE(user_events, event_time, INTERVAL '1 DAY')
WHERE event_type = 'login'
GROUP BY window_start;
CREATE MATERIALIZED VIEW mau_rolling_30d AS
SELECT
window_end::DATE AS snapshot_date,
COUNT(DISTINCT user_id) AS mau
FROM HOP(user_events, event_time, INTERVAL '1 DAY', INTERVAL '30 DAYS')
WHERE event_type = 'login'
GROUP BY window_end;
CREATE MATERIALIZED VIEW stickiness AS
SELECT d.activity_date, d.dau, m.mau,
ROUND(d.dau::NUMERIC / NULLIF(m.mau, 0) * 100, 2) AS stickiness_pct
FROM dau d
JOIN mau_rolling_30d m ON d.activity_date = m.snapshot_date;
Store event_time as TIMESTAMPTZ and apply AT TIME ZONE inside the window if day boundaries should follow your primary market. If producers emit duplicate logins after reconnects, deduplicate on (user_id, event_time) in a view before the count.
Audience Segments, Retargeting, and Suppression
Segment membership is a continuously updated query result rather than a nightly snapshot. The same event stream feeds high-intent segments for retargeting, interest segments for personalization, and suppression lists that remove recent converters within seconds of purchase so you stop paying to retarget people who already bought.
A seven-day hopping window that slides every 30 minutes classifies each user by the deepest funnel stage reached, and a join against recent purchasers sets the suppression flag:
CREATE MATERIALIZED VIEW retargeting_audiences AS
WITH activity AS (
SELECT
user_id,
window_start,
window_end,
COUNT(*) FILTER (WHERE event_type = 'product_view') AS product_views,
COUNT(*) FILTER (WHERE event_type = 'add_to_cart') AS cart_adds,
COUNT(*) FILTER (WHERE event_type = 'checkout_start') AS checkout_starts,
MAX(value) FILTER (WHERE event_type = 'product_view') AS max_product_value,
MAX(event_time) AS last_active
FROM HOP(user_events, event_time, INTERVAL '30 MINUTES', INTERVAL '7 DAYS')
GROUP BY user_id, window_start, window_end
),
converters AS (
SELECT DISTINCT user_id
FROM user_events
WHERE event_type = 'purchase' AND event_time >= NOW() - INTERVAL '7 DAYS'
)
SELECT
a.*,
CASE
WHEN a.checkout_starts > 0 THEN 'cart_abandoner'
WHEN a.cart_adds > 0 THEN 'cart_viewer'
WHEN a.product_views >= 3 THEN 'high_intent_browser'
WHEN a.product_views >= 1 THEN 'product_viewer'
ELSE 'site_visitor'
END AS audience_segment,
(c.user_id IS NOT NULL) AS recently_converted
FROM activity a
LEFT JOIN converters c USING (user_id);
Session windows add category-level interest for personalization, and a filtered session view catches visits where the viewed product value crossed a threshold:
CREATE MATERIALIZED VIEW category_interest_sessions AS
SELECT user_id, category, window_start, window_end,
COUNT(*) AS views_in_session, SUM(value) AS viewed_value
FROM SESSION(user_events, event_time, INTERVAL '30 MINUTES')
WHERE event_type = 'product_view'
GROUP BY user_id, category, window_start, window_end
HAVING COUNT(*) >= 2;
CREATE MATERIALIZED VIEW high_value_sessions AS
SELECT user_id, window_start, window_end, COUNT(*) AS events_in_session, SUM(value) AS session_value
FROM SESSION(user_events, event_time, INTERVAL '30 MINUTES')
GROUP BY user_id, window_start, window_end
HAVING SUM(value) > 100;
Export the audience view to Kafka as an upsert stream. A small consumer calls your ad platform's Customer Match or Custom Audience API, adding users when they enter a segment and removing them when recently_converted flips to true:
CREATE SINK retargeting_audience_export FROM retargeting_audiences
WITH (connector = 'kafka', properties.bootstrap.server = 'kafka:9092', topic = 'retargeting.audiences.updates')
FORMAT UPSERT ENCODE JSON;
Behavioral Email Triggers with Cooldowns
Most marketing automation platforms trigger emails through polling loops every 5 to 15 minutes. RisingWave replaces the loop with views that detect the condition as events arrive. RisingWave decides when to trigger; Braze, Iterable, Postmark, or SendGrid decides what to send.
Cart abandonment uses a session window with a HAVING clause: the session added to cart and did not purchase. Browse abandonment uses an hourly window: three or more product views and no cart add.
CREATE MATERIALIZED VIEW cart_abandonment_triggers AS
SELECT
user_id,
window_start AS session_start,
window_end AS session_end,
MAX(value) FILTER (WHERE event_type = 'add_to_cart') AS abandoned_cart_value,
COUNT(*) FILTER (WHERE event_type = 'add_to_cart') AS items_added
FROM SESSION(user_events, event_time, INTERVAL '30 MINUTES')
GROUP BY user_id, window_start, window_end
HAVING COUNT(*) FILTER (WHERE event_type = 'add_to_cart') > 0
AND COUNT(*) FILTER (WHERE event_type = 'purchase') = 0;
CREATE MATERIALIZED VIEW browse_abandonment_triggers AS
SELECT user_id, window_start, window_end,
COUNT(*) FILTER (WHERE event_type = 'product_view') AS products_viewed,
COUNT(DISTINCT product_id) AS unique_products
FROM TUMBLE(user_events, event_time, INTERVAL '1 HOUR')
GROUP BY user_id, window_start, window_end
HAVING COUNT(*) FILTER (WHERE event_type = 'product_view') >= 3
AND COUNT(*) FILTER (WHERE event_type = 'add_to_cart') = 0;
Frequency control matters as much as timing. Stream the email platform's send log back into RisingWave and join it, together with a suppression table, before anything reaches the sink:
CREATE SOURCE email_send_events (
send_id VARCHAR,
user_id VARCHAR,
trigger_type VARCHAR,
sent_at TIMESTAMPTZ
) WITH (
connector = 'kafka',
topic = 'email-sends',
properties.bootstrap.server = 'kafka:9092',
scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;
CREATE TABLE email_suppression (user_id VARCHAR PRIMARY KEY, reason VARCHAR, suppressed_at TIMESTAMPTZ);
CREATE MATERIALIZED VIEW cart_abandonment_sendable AS
SELECT t.*
FROM cart_abandonment_triggers t
LEFT JOIN email_suppression s USING (user_id)
LEFT JOIN (
SELECT user_id FROM email_send_events
WHERE trigger_type = 'cart_abandonment' AND sent_at >= NOW() - INTERVAL '24 HOURS'
) recent USING (user_id)
WHERE s.user_id IS NULL
AND recent.user_id IS NULL;
CREATE SINK cart_abandonment_email_sink FROM cart_abandonment_sendable
WITH (connector = 'kafka', topic = 'email-triggers-cart-abandonment', properties.bootstrap.server = 'kafka:9092')
FORMAT PLAIN ENCODE JSON;
Serving Everything
Dashboards connect to RisingWave over PostgreSQL on port 4566 and query the views directly. For systems that expect their own tables, upsert sinks keep them current:
CREATE SINK customer_clv_sink FROM customer_clv
WITH (connector = 'jdbc', jdbc.url = 'jdbc:postgresql://app-db:5432/crm',
table.name = 'customer_clv_live', type = 'upsert', primary_key = 'customer_id');
CREATE SINK attribution_to_warehouse FROM channel_attribution_daily
WITH (connector = 'iceberg', type = 'upsert', catalog.type = 'glue',
s3.bucket = 'your-data-lake-bucket', database.name = 'marketing', table.name = 'channel_daily_attribution',
primary_key = 'channel,campaign_id,window_start');
Kafka sinks carry A/B conversion rows to a rollback service, audience updates to the ad platform connector, and email triggers to the sending platform. Iceberg sinks archive funnel and attribution history for quarter-over-quarter analysis.
Batch Marketing Analytics vs Streaming SQL
| Capability | Nightly warehouse rollup | Polling automation platform | Streaming SQL (RisingWave) |
|---|---|---|---|
| Funnel and pacing refresh | Hours to a day | Not covered | Per event |
| A/B rollback signal | Next morning | Not covered | Minutes, via rolling funnel |
| Attribution model | Overnight job, one model | Vendor model | Continuous, several models side by side |
| Lifetime value and tiers | Daily snapshot | Not covered | Updates on every order |
| DAU, MAU, stickiness | Daily | Not covered | Daily tumble plus 30-day hop, live |
| Segment membership | Hourly or daily export | 5 to 15 minute poll | Seconds, with converter suppression |
| Trigger timing | Hours | 5 to 15 minutes | Seconds after the session closes |
| Pipeline code | SQL plus scheduler plus orchestrator | Vendor UI | SQL only |
Frequently Asked Questions
Can I use one event stream for all of these views?
Yes, and it is the recommended design. Every view in this guide reads from user_events, touchpoint_events, or orders. Adding a new metric is a new CREATE MATERIALIZED VIEW, not a new pipeline.
How do I handle identity across devices?
Resolve identity before the event reaches Kafka where you can, so user_id is already unified. Where you cannot, stream your identity graph table into RisingWave through CDC and join it in a view that rewrites anonymous IDs to the resolved ID before the aggregations run.
What lookback window should attribution use? Thirty days is the common default and the one used above. Make it a column in a configuration table and join on it if channels need different windows, such as one day for display impressions and 30 days for paid search clicks.
Does the rolling A/B funnel double count users across overlapping windows?
Each window counts distinct users within that window only. Overlap is intentional: it is what makes the trend smooth. Use ab_variant_conversion for lifetime totals and ab_hourly_funnel for the current trend.
How do I backfill history when I first deploy?
Set scan.startup.mode = 'earliest' on the Kafka sources. RisingWave processes retained history first and then continues streaming, so CLV, DAU, and attribution views start with full history rather than from zero.
