[docs] Add Real-Time User Profile quickstart tutorial - #2669
Conversation
|
Hi @wuchong PTAL! |
wuchong
left a comment
wuchong
left a comment
There was a problem hiding this comment.
Hi @Prajwal-banakar, thank you for your contribution! However, quickstart documentation typically needs to be fully reproducible—readers should be able to follow it step by step and achieve the same results, just like in our existing quickstarts:
https://fluss.apache.org/docs/quickstart/flink/ and https://fluss.apache.org/docs/quickstart/lakehouse/.
Could you please enhance the guide by adding the environment setup (e.g., using Docker Compose), clear instructions on how to run the queries, and guidance on how to visualize or verify the results? This will greatly improve usability and consistency with our documentation standards.
|
Hi @Prajwal-banakar thanks for the pr. I have the same suggestion as @wuchong . We need to make the example fully reproducible. For example, to ingest the raw data for the source datastream, we can provide a csv file as sample data, just like this example: https://github.com/aliyun/alibabacloud-hologres-connectors/blob/master/hologres-connector-examples/hologres-connector-flink-examples/src/main/java/com/alibaba/hologres/connector/flink/example/FlinkRoaringBitmapAggJob.java |
|
Hi @Prajwal-banakar, please ensure the quickstart can run successfully. Additionally, the image appears to be AI-generated. I don’t object to AI-generated content in principle, but please make sure the text in the image contains no garbled characters and all content makes sense. |
|
Hi @wuchong verified the guide locally it is working ! and fixed the diagram format, |
| d.uid, | ||
| -- Convert INT to BYTES for rbm64. | ||
| -- Note: In a real production job, you might use a UDF to ensure correct bitmap initialization. | ||
| CAST(CAST(d.uid AS STRING) AS BYTES), |
There was a problem hiding this comment.
I believe we need the to_rbm and from_rbm Flink UDFs to process the data correctly. Without these functions, the results would be meaningless and users would not understand the purpose of this feature.
However, shipping Flink UDFs falls outside the scope of the Fluss project. I will coordinate with members of the Flink community to contribute these UDFs and identify an appropriate location to open source and publish the UDF JARs. Once available, we can reference these functions in our documentation and examples.
That said, the Lunar New Year holiday is approaching in China, so we likely will not be able to start this work until March. Until then, we may need to suspend this PR. Thank you for your quick updating.
There was a problem hiding this comment.
I believe we need the
to_rbmandfrom_rbmFlink UDFs to process the data correctly. Without these functions, the results would be meaningless and users would not understand the purpose of this feature.
Exactly. we need such functions as RB_CARDINALITY and RB_OR_AGG for aggregating the result bitmap.
|
Hi @wuchong, @xx789633, |
|
Hi @wuchong @xx789633 i've created a temporary repo for flink udfs as we discussed in slack and updated the quick start guide, it is working smoothly. The repo is available at : https://github.com/Prajwal-banakar/flink-roaringbitmap 
|
|
Hi @Prajwal-banakar , |
|
Hi @platinumhamburg, thank you for the kind words! I will move the RoaringBitmap UDFs to that repo and i'll open a PR their shortly. I also have an active [DISCUSS] thread on the mailing list proposing Native Bitmap Integration for Fluss https://lists.apache.org/thread/z9dwyg81cs3bt7yssb4n3vg17o767r5s and it is open for your input, so I am happy to contribute UDFs there as well and collaborate on expanding the library. I will update this PR once the JAR is published from the official repo. |
|
Hi @platinumhamburg, I tried to fork https://github.com/flink-extended/flink-roaringbitmap |
Hi @Prajwal-banakar, sorry about that—the reason you couldn’t fork earlier was that the repository was empty. It’s now ready to go. Also, could we add an rb_or_agg function in the initial implementation? This would help us include a roll-up query example in the quick start guide. |
|
Hello @Prajwal-banakar.. Thank you for this, it's really nice work. However I have the two following comments:
|
|
Hi @polyzos, thank you for the review! Regarding point 1: The rbm64 aggregation type is already officially supported in Fluss 0.9 (FIP-21). The companion UDFs (rb_build_agg, rb_cardinality) are now publishing in the official flink-extended/flink-roaringbitmap repo, which was set up by @wuchong and @platinumhamburg specifically to support this quickstart. So the dependency is intentional and endorsed by the maintainers. Regarding point 2: Completely agree. I'll replace the diagram with a clean one made in Excalidraw. |
|
@Prajwal-banakar correct and apologies I should have been more precise. What I was thinking is to have the quickstart work with Fluss specific features in the quickstart to keep the complexity to the bare minimum for new users - i.e how they can use the auto-increment column along with the aggregation merge engine. Then create an "extended" version in a blog post for those looking for more.. This approach will help as I mentioned:
This is just a suggestion, both approaches I think are valuable and I'm happy with both.. Just cautious for new users., that's all. |
|
Hi @polyzos, that makes complete sense and I appreciate the clarification! I agree with this approach:
I'll simplify this PR to the basic quickstart version and work on the blog post separately. @wuchong @platinumhamburg does this approach work for you as well? |
|
Hi @wuchong @polyzos @platinumhamburg, Quick update: the external RoaringBitmap UDF library has now been officially released as That said, I agree with the latest direction for this PR keeping the Fluss quickstart focused on a minimal, Fluss-native path will make it much easier for new users to follow. I’ll update this PR accordingly by simplifying it to focus on the auto-increment column + aggregation merge engine workflow, and @polyzos is working on a blog/deep-dive covering the extended bitmap use case. Please let me know if there’s any specific structure or example you’d prefer for the simplified quickstart. |
|
@Prajwal-banakar looking forward to the update.. you can also take a look here |
| uid INT, | ||
| PRIMARY KEY (email) NOT ENFORCED | ||
| ) WITH ( | ||
| 'connector' = 'fluss', |
There was a problem hiding this comment.
since we are already in the catalog with dont need to specify the connector for every CREATE TABLE statement
| ) WITH ( | ||
| 'connector' = 'fluss', | ||
| 'auto-increment.fields' = 'uid', | ||
| 'bucket.num' = '1' |
There was a problem hiding this comment.
bucket.num = 1 is the default for every table, so again we can remove this for every CREATE TABLE statement.
|
can we also change the name from |
|
Hi @polyzos, I’ve updated the PR and verified locally it is working fine. |
platinumhamburg
left a comment
platinumhamburg
left a comment
There was a problem hiding this comment.
Thanks @Prajwal-banakar , I left some comments.
| SELECT profile_id, total_clicks FROM user_profiles; | ||
| ``` | ||
|
|
||
| You should see 5 rows (one per profile group) with `total_clicks` increasing in real time as new events arrive. |
There was a problem hiding this comment.
'fields.profile_group_id.expression' = '#{number.numberBetween ''1'',''5''}'
Java Faker's numberBetween(min, max) treats max as exclusive. This generates values 1–4, not 1–5. Should be ''1'',''6'' to produce 5 distinct profile groups. Same issue on line 162: numberBetween '1','10' generates 1–9, not 1–10 — change to ''1'',''11'' if 10 is intended.
| SELECT | ||
| e.profile_group_id, | ||
| CAST(e.click_count AS BIGINT) | ||
| FROM raw_events AS e |
There was a problem hiding this comment.
The uid column from user_dict is never referenced in the SELECT. The lookup join exists solely to trigger insert-if-not-exists, but the generated ID plays no role in the aggregation. This makes the pipeline feel contrived — a reader would expect the dictionary-mapped uid to be the primary key of user_profiles, not an unrelated profile_group_id.
|
Hi @platinumhamburg, thanks for comments, fixed both issues, PTAL! |
|
@Prajwal-banakar can you also rebase on master? t fix the doc failing issue |
32e6662 to
4a8a93c
Compare
|
Hi @polyzos, the docs failure was caused by the blog reference in the “What’s Next” section before that PR is merged. I removed the link, rebased |
|
Hi @wuchong @polyzos @platinumhamburg this PR was previously blocked because it referenced external UDF JARs for bitmap operations. Now that FIP-37 has landed and rb_build_agg, rb_cardinality, and the other bitmap functions are natively available in FlussCatalog, can we update this PR to use the native functions instead of the external JAR approach or would you prefer this becomes a standalone quickstart that references the new SQL Functions docs page? Happy to go either direction. |
|
Thank you @pbanakar , I think we should update it, it would be great to include it in the 1.0 release. |
|
Hi @wuchong @polyzos @platinumhamburg, Quick question : Since FIP-37 functions (rb_build_agg, rb_cardinality, etc.) are targeting the 1.0 release and the Docker image is not yet available, I cannot test the bitmap version locally right now. The current PR uses sum-only aggregation with 0.9-incubating and has been fully tested end-to-end. I'm happy to update it to use the native bitmap functions once 1.0 is released and we can verify it locally. i think we have 2 options now:
Happy to go either direction. WDYT? |
4a8a93c to
d8a0782
Compare
@pbanakar Thanks for continuing to improve this quickstart. Given its original goal—to demonstrate a realistic user-profile workflow combining auto-increment IDs with the aggregation merge engine—and since the earlier bitmap dependency is now addressed by the native FIP-37 functions, I would lean toward option 2. |
|
Hi @platinumhamburg, that makes sense! I'll build from the current main branch locally to test the bitmap version, then do a final verification with the 1.0 RC before merging. One quick question, what should I use as the FLUSS_DOCKER_VERSION in the tutorial for now? Should it reference main-SNAPSHOT or wait until the 1.0 RC tag is available? |
Hi @pbanakar, please use the existing documentation placeholder |
|
Thank you @platinumhamburg, i'm greatefull for your guidence, I tested the bitmap quickstart locally using the main branch build and hit a runtime issue I want to flag. Error: What I tried:
Root cause hypothesis: i think the suggested fix is Either include RoaringBitmap in prepare_build.sh with the correct unshaded version that matches what Fluss expects, or ensure the Fluss connector does not shade RoaringBitmap so the external JAR resolves correctly!? Happy to open a separate issue for this if helpful. PTAL! |
@pbanakar Thanks for testing this and for reporting the runtime issue. The |
|
Thanks @platinumhamburg! I found a fix and ran the full quickstart locally after applying #3981 everything is working fine now, and PTAL on that PR if you have some time and I'll update this PR tonight to use the RB functions, Thanks again for your guidance! 🙏 |
There was a problem hiding this comment.
Hi @pbanakar ,
The current data model should be reworked rather than only adjusting the individual SQL statements. The table is currently keyed by uid, while the bitmap element is also built from the same uid. For a row keyed by uid = u, every incoming bitmap is therefore {u}, so the rbm32 union remains {u} and rb_cardinality can never grow beyond 1. This mixes two different grains: a per-user profile and a business-dimension UV aggregation.
The current GROUP BY d.uid, e.click_count also introduces an unbounded Flink aggregation keyed by a high-cardinality UID. In addition to contradicting the stateless-Flink claim, repeated events with the same (uid, click_count) are collapsed into the same aggregation group instead of contributing independently to the click total.
I suggest changing this quickstart to a page-view user-profile use case and renaming it to page_user_profile. The table should store hourly page-view statistics, following the model described in Real-Time Multi-Dimensional Unique Visitor Deduplication in Practice: the primary key represents the business dimensions and time bucket, while uid is only the visitor identifier stored in the bitmap.
The dictionary table can retain the current email -> uid mapping:
CREATE TABLE user_dict (
email STRING,
uid INT,
PRIMARY KEY (email) NOT ENFORCED
) WITH (
'auto-increment.fields' = 'uid'
);The aggregation table can then be defined as follows:
CREATE TABLE page_user_profile (
channel STRING,
city STRING,
ymd STRING,
hh STRING,
uv_bitmap BYTES,
pv BIGINT,
PRIMARY KEY (channel, city, ymd, hh) NOT ENFORCED
) WITH (
'table.merge-engine' = 'aggregation',
'fields.uv_bitmap.agg' = 'rbm32',
'fields.pv.agg' = 'sum'
);This table stores one cumulative bitmap and one PV counter for each (channel, city, ymd, hh) combination. If the tutorial is intended to calculate statistics for each individual page, page_id should also be included in the source and primary key.
The Faker source should also be adapted to the new model:
- Emails should come from a bounded but sufficiently large population so that users repeat and demonstrate deduplication without every channel immediately reaching the maximum UV.
channelandcityshould come from finite enumerations.- Event timestamps should cover the most recent 36 hours, with
ymdandhhderived from the generated timestamp. - Each generated row should represent one page view, so its PV contribution is always 1.
With 10 rows per second, a population of approximately 500 email identities is a reasonable balance. After one minute, each of the three channels receives approximately 200 page views and is expected to have roughly 165 unique users. After three minutes, each channel receives approximately 600 page views and has roughly 350 unique users. This makes the UV < PV effect visible without immediately saturating every channel at the maximum user count.
For example:
CREATE TEMPORARY TABLE page_views (
email STRING,
channel STRING,
city STRING,
event_time TIMESTAMP(3),
ymd AS DATE_FORMAT(event_time, 'yyyyMMdd'),
hh AS DATE_FORMAT(event_time, 'HH'),
proctime AS PROCTIME()
) WITH (
'connector' = 'faker',
'rows-per-second' = '10',
-- Generate 500 email addresses:
-- user_000@example.com through user_499@example.com.
'fields.email.expression' =
'#{regexify ''user_[0-4][0-9][0-9]@example[.]com''}',
'fields.channel.expression' =
'#{Options.option ''app'',''web'',''mini_program''}',
'fields.city.expression' =
'#{Options.option ''Amsterdam'',''Berlin'',''New York''}',
-- Distribute events across the most recent 36 hours.
'fields.event_time.expression' =
'#{date.past ''36'',''HOURS''}'
);Because the events are distributed across:
3 channels × 3 cities × 36 hours = 324 hourly buckets
the UV and PV of an individual hourly row may initially be close to each other. The more visible deduplication effect will appear when these hourly bitmaps are rolled up by channel, city, date, or another dimension in the OLAP stage. This is also the intended benefit of retaining composable bitmaps rather than materializing only their cardinalities.
The write pipeline should emit one singleton bitmap and one PV increment for every page-view event. It should not use rb_build_agg or a non-windowed GROUP BY in Flink:
INSERT INTO page_user_profile
SELECT
e.channel,
e.city,
e.ymd,
e.hh,
rb_build(ARRAY[d.uid]) AS uv_bitmap,
CAST(1 AS BIGINT) AS pv
FROM page_views AS e
JOIN user_dict
/*+ OPTIONS('lookup.insert-if-not-exists' = 'true') */
FOR SYSTEM_TIME AS OF e.proctime AS d
ON e.email = d.email;This keeps Flink responsible only for identity mapping and event forwarding. Fluss performs the hourly bitmap union and PV summation in the Aggregation Merge Engine without indefinitely growing Flink group-aggregation state.
At the OLAP stage, PV can be summed directly, but UV cardinalities must not be summed across hourly rows. The bitmaps must first be unioned so that a user appearing in multiple cities, channels, or hours is counted only once.
For example, roll up the latest 36-hour data by channel:
SELECT
channel,
rb_cardinality(rb_or_agg(uv_bitmap)) AS uv,
SUM(pv) AS pv
FROM page_user_profile
GROUP BY channel;Roll up by city:
SELECT
city,
rb_cardinality(rb_or_agg(uv_bitmap)) AS uv,
SUM(pv) AS pv
FROM page_user_profile
GROUP BY city;A daily roll-up by channel can retain ymd:
SELECT
channel,
ymd,
rb_cardinality(rb_or_agg(uv_bitmap)) AS uv,
SUM(pv) AS pv
FROM page_user_profile
GROUP BY channel, ymd;Other combinations, such as channel + city, can use the same pattern by changing the GROUP BY dimensions. This demonstrates the main value of storing hourly RoaringBitmaps: the OLAP layer can compose them across arbitrary dimensions and time ranges without double-counting users.
Since the Faker source is unbounded, any finite email population will eventually appear in every channel if the job runs long enough. The 500-email population is intended to provide a useful quickstart demonstration window at 10 rows per second rather than model an indefinitely running production workload.
| " | ||
| ``` | ||
|
|
||
| ## Step 1: Create the Fluss Catalog |
There was a problem hiding this comment.
All steps should use H3 to make them under the Enter the SQL Client section.
| Use the following command to enter the Flink SQL Client: | ||
|
|
||
| ```shell | ||
| docker compose run --entrypoint bash sql-client -c " |
There was a problem hiding this comment.
Can we simplify this to docker compose run sql-client, as in the existing Flink quickstart? The service already sets command: ["/opt/sql-client/sql-client"], and the required REST settings are provided through FLINK_PROPERTIES, so the custom entrypoint and duplicated arguments are unnecessary. Since this tutorial creates its own source table, we may also want to point the service command at /opt/flink/bin/sql-client.sh to avoid preloading the unrelated demo tables from sql-client.sql.
|
|
||
| ## Step 5: Verify Results | ||
|
|
||
| Open a **second terminal**, navigate to the working directory, and launch another SQL Client session to query results while the pipeline runs. |
There was a problem hiding this comment.
Why do we need a second terminal here? Flink SQL Client executes DML asynchronously by default (table.dml-sync=false), so the prompt becomes available again after the INSERT INTO job is submitted. Since this tutorial never enables synchronous DML, we can run the verification queries in the same session and remove the duplicated SQL Client startup and catalog setup.
| SELECT | ||
| d.uid, | ||
| CAST(e.click_count AS BIGINT), | ||
| rb_build_agg(d.uid) |
There was a problem hiding this comment.
The user_profiles row key and the bitmap element are both d.uid. For a row keyed by UID u, every incoming bitmap is therefore {u}, so the rbm32 union remains {u} and rb_cardinality(unique_visitors) can never grow beyond 1. This does not demonstrate unique visitors as the tutorial claims. Please introduce a separate aggregation key, such as a page, campaign, or profile-group ID, as the table primary key and keep d.uid as the visitor ID stored in the bitmap.
| JOIN user_dict /*+ OPTIONS('lookup.insert-if-not-exists' = 'true') */ | ||
| FOR SYSTEM_TIME AS OF e.proctime AS d | ||
| ON e.email = d.email | ||
| GROUP BY d.uid, e.click_count; |
There was a problem hiding this comment.
Using rb_build_agg here introduces an unbounded Flink group aggregation keyed by (uid, click_count). Since uid is high-cardinality and these groups never close, Flink must retain continuously growing keyed state, which defeats the purpose of offloading aggregation to the Fluss Aggregation Merge Engine. We can remove the GROUP BY and use the scalar rb_build(ARRAY[d.uid]) to emit a singleton bitmap for each event; the Fluss rbm32 field will union these bitmaps, while sum aggregates the click counts on the server side.
5 participants
Purpose
Linked issue: close #2659
The purpose of this change is to add a new quickstart tutorial, "Real-Time User Profile," to the Apache Fluss documentation. This tutorial demonstrates a realistic, production-grade business scenario by combining the Auto-Increment Column and Aggregation Merge Engine features. It specifically addresses the need for guidance on mapping high-cardinality string identifiers to compact integers for efficient real-time analytics.
Brief change log
This pull request introduces a comprehensive tutorial located at website/docs/quickstartUuser-Profile.md. Key changes include:
Realistic Use Case: Developed a scenario focused on identity mapping (Email to UID) and real-time metric aggregation (Total Clicks and Unique Visitors).
Feature Integration: Showcases the synergy between FIP-16 (Auto-Increment) for dictionary management and FIP-21 (Aggregation Merge Engine) for storage-level pre-aggregation.
Technical Optimization: Implemented the maintainer's recommendation to use INT for the generated uid column to maximize storage efficiency and performance for RoaringBitmap (rbm64) operations.
Reliability Section: Added documentation on Undo Recovery to explain how Fluss ensures exactly-once accuracy for aggregations during Flink failovers.
Visual Guidance: Included an architectural diagram to illustrate the data flow from raw event ingestion to the final pre-aggregated profile storage.
Tests
Documentation Build: Verified that the documentation builds correctly using the local Docusaurus environment and that the new page is correctly linked in the sidebar.
SQL Verification: Manually verified the Flink SQL syntax against the Apache Fluss 0.9 connector specifications.
API and Format
This change is documentation-only and does not affect the Java API or the underlying storage format.
Documentation
Yes, this change introduces a new documentation feature (a new quickstart tutorial) aimed at guiding users through the adoption of Fluss's advanced streaming storage capabilities.