Data Platform Analytics — dbt Playbook¶

Hướng dẫn phát triển analytics trên data platform sử dụng dbt (streaming + batch).

Prerequisites¶

1. Install dbt locally¶

pip install dbt-core==1.9.0 dbt-risingwave==1.9.0 dbt-trino==1.9.0

Verify:

dbt --version
# dbt-core: 1.9.0
# dbt-risingwave: 1.9.0
# dbt-trino: 1.9.0

2. Data platform must be running¶

scripts/dp-check.sh
# Must show: ✅ DATA PLATFORM HEALTHY (27/27 passed)

Architecture¶

graph TD
    PUSH["PUSH API :8093
(sole data entry)"] --> RP[Redpanda Kafka
tmpfs 2GB]
    RP --> RW["RisingWave v3.0.0 :4566
single_node persistent (8GB RAM)"]
    RP -.->|via Iceberg sinks| TR

    subgraph DBT["dbt-risingwave (analytics/streaming)"]
        STG["staging/
stg_chat_messages (COALESCE field mapping)
stg_chat_messages_valuable (DM + owner filter)"]
        MV["mvs/
5 materialized views"]
        SNK["sinks/
Iceberg sink to Lakekeeper"]
        STG --> MV
        MV --> SNK
    end
    RW <-.->|dbt-risingwave| DBT

    TR["Trino :8091
dbt-trino (batch models)
catalog=gravitino_iceberg (legacy name)
→ Lakekeeper :8181 REST catalog
warehouse=homelab"]
    LK["Lakekeeper :8181
Iceberg REST catalog"] --> RFS["RustFS s3://iceberg/homelab/"]
    TR --> LK
    LK --> RFS

    RW --> SS[Superset BI]
    TR --> SS

Two dbt projects¶

Important: One dbt run targets ONE adapter. You cannot mix RW + Trino models in a single dbt run.

Directory structure¶

analytics/
├── streaming/                      ← dbt-risingwave project
│   ├── dbt_project.yml             # Project config (materialized_view default)
│   ├── profiles.yml                # RW connection (env vars)
│   └── models/
│       ├── sources.yml             # Declare existing Kafka sources (external)
│       ├── staging/                # 3-layer staging: clean raw Kafka data
│       │   ├── stg_chat_messages.sql            # COALESCE field mapping
│       │   └── stg_chat_messages_valuable.sql   # DM + owner filter + thread_category
│       ├── mvs/                    # Materialized views (consume staging)
│       │   ├── mv_daily_activity.sql
│       │   ├── mv_hourly_heatmap.sql
│       │   ├── mv_events_hourly.sql
│       │   ├── mv_metrics_5min.sql
│       │   └── mv_logs_hourly.sql
│       └── sinks/                  # Iceberg sinks (RW → Lakekeeper)
│           └── iceberg_logs_sink.sql
│
├── batch/                          ← dbt-trino project
│   ├── dbt_project.yml             # Project config (view + incremental default)
│   ├── profiles.yml                # Trino connection (env vars)
│   └── models/
│       ├── sources.yml             # Declare Iceberg tables (via Lakekeeper REST)
│       ├── staging/                # Thin wrappers
│       │   └── stg_platform_logs.sql
│       └── marts/                  # Final analytics tables
│           └── fct_daily_log_summary.sql

Development workflow¶

Step 1: Edit SQL¶

# Add a new streaming MV
vim analytics/streaming/models/mvs/mv_top_senders.sql

{{
  config(
    materialized='materialized_view'
  )
}}

-- Top 10 most active senders
SELECT
    sender_id,
    sender_name,
    thread_name,
    count(*) AS message_count
FROM {{ source('kafka', 'chat_messages') }}
GROUP BY sender_id, sender_name, thread_name
ORDER BY message_count DESC
LIMIT 10

Step 2: Test locally¶

cd analytics/streaming

# Check dbt can parse
dbt parse --profiles-dir .

# Dry-run: compile SQL without executing
dbt compile --profiles-dir .

# Apply
dbt run --profiles-dir .

# Verify in RW
psql -h 100.126.172.96 -p 4566 -U root -d dev -c \
  "SELECT * FROM mv_top_senders LIMIT 5"

Step 3: Commit + push¶

cd /workspaces/github.com/minhluc-info/docker-compose
git add analytics/
git commit -m "feat(analytics): add mv_top_senders"
git push

Step 4: Deploy¶

scripts/deploy-dbt.sh streaming    # or: batch, all

Common tasks¶

Add a new streaming MV¶

1. Create analytics/streaming/models/mvs/mv_name.sql
2. Use {{ source('kafka', 'source_name') }} to reference sources
3. Set materialized='materialized_view' in config
4. Run: dbt run --select mv_name --profiles-dir analytics/streaming/

Add a new batch model¶

1. Create analytics/batch/models/marts/fct_name.sql
2. Use {{ source('iceberg', 'table_name') }} for Iceberg sources
3. Use {{ ref('stg_name') }} for staging model dependencies
4. Set materialized='incremental' with unique_key for incremental loads
5. Run: dbt build --select fct_name --profiles-dir analytics/batch/

Add a new source¶

Streaming (RisingWave): Sources are Kafka tables created by risingwave-init.sql. Declare them in sources.yml:

# analytics/streaming/models/sources.yml
sources:
  - name: kafka
    schema: public
    tables:
      - name: new_source_table # Must already exist in RW via CREATE SOURCE

Batch (Trino): Sources are Iceberg tables registered via Lakekeeper REST catalog. Declare in sources.yml:

# analytics/batch/models/sources.yml
sources:
  - name: iceberg
    schema: dev_streaming
    tables:
      - name: new_iceberg_table # Must exist in Lakekeeper warehouse=homelab

Run dbt tests¶

cd analytics/streaming
dbt test --profiles-dir .

Generate documentation¶

cd analytics/streaming
dbt docs generate --profiles-dir .
dbt docs serve --profiles-dir .  # Open http://localhost:8081

Staging layer (3-layer model)¶

The streaming project uses a 3-layer model (staging → mvs → sinks). The staging layer unifies field names across heterogeneous producers and filters noisy data before downstream MVs aggregate it.

stg_chat_messages — field mapping via COALESCE¶

Different producers emit different field names for the same logical concept. stg_chat_messages normalizes them using COALESCE (picks whichever side is populated):

Common fields (provider, thread_id, thread_name, thread_type, text, media_type, reply_to_id, metadata) pass through unchanged.

Downstream MVs MUST source from stg_chat_messages (not the raw Kafka table) so they automatically support both Telegram and Push API producers.

stg_chat_messages_valuable — DM + owner filter¶

Filters out noise (spam, group chatter with no owner participation) and adds a thread_category classification column for downstream segmentation.

Keep rules:

Owner identification per provider:

• Zalo: msg_sender_name = 'Bạn' (Vietnamese for "you" — Zalo's convention for self)
• Telegram: all messages kept (already curated via WHITELIST_THREAD_IDS in dp-chat-adapter)
• Others: all kept

Spam keywords (substring match, case-insensitive):

tuyển dụng, tuyển gấp, công nhật, khuyến mãi, giảm giá, mua 1 tặng

Using staging in new MVs¶

-- ✅ GOOD — consume staging (handles both producers + filters noise)
SELECT count(*) FROM {{ ref('stg_chat_messages_valuable') }}
WHERE thread_category IN ('dm', 'group_owner')

-- ❌ BAD — bypasses staging, breaks on Push API rows
SELECT count(*) FROM {{ source('kafka', 'chat_messages') }}

Connection details¶

RisingWave (streaming)¶

# analytics/streaming/profiles.yml
streaming:
  target: dev
  outputs:
    dev:
      type: risingwave
      host: 100.126.172.96 # Tailscale IP
      port: 4566
      user: root
      password: ""
      dbname: dev
      schema: public

Trino (batch)¶

# analytics/batch/profiles.yml
batch:
  target: dev
  outputs:
    dev:
      type: trino
      host: 100.126.172.96 # Tailscale IP
      port: 8091
      user: user
      catalog: gravitino_iceberg # legacy name — actually points to Lakekeeper :8181
      schema: analytics # Trino will create this schema

Lakekeeper requirement (MANDATORY): The Trino catalog gravitino_iceberg is configured in 30-query/compose.yaml with:

connector.name=iceberg
iceberg.catalog.type=rest
iceberg.rest-catalog.uri=http://192.168.100.31:8181/catalog
iceberg.rest-catalog.warehouse=homelab
iceberg.rest-catalog.nested-namespace-enabled=false
fs.native-s3.enabled=true
s3.endpoint=http://192.168.100.59:9000
s3.path-style-access=true

The warehouse=homelab value MUST match a warehouse created in Lakekeeper (see data-platform-bootstrap.md Phase 4). If the warehouse doesn't exist, every Trino query fails with Requested warehouse does not exist. The catalog filename (gravitino_iceberg.properties) is kept for legacy compatibility — it does NOT mean Gravitino is in use.

Debug guide¶

"dbt: command not found"¶

pip install dbt-core==1.9.0 dbt-risingwave==1.9.0 dbt-trino==1.9.0

"Could not find adapter risingwave"¶

dbt-risingwave not installed or version mismatch:

pip install dbt-risingwave==1.9.0
dbt --version  # verify adapter listed

"Connection refused" to RisingWave¶

# Check RW is running
scripts/dp-check.sh

# Test connection manually
psql -h 100.126.172.96 -p 4566 -U root -d dev -c "SELECT 1"

"Connection refused" to Trino¶

# Trino needs ~2 min to start after deploy
curl http://100.126.172.96:8091/v1/info
# Must show: "starting": false

# If still starting, wait and retry
sleep 60 && curl http://100.126.172.96:8091/v1/info

"source 'kafka' not found"¶

Source declaration missing in sources.yml. Add the source name:

sources:
  - name: kafka
    schema: public
    tables:
      - name: your_source_name # Must match table name in RW

"source 'iceberg' not found"¶

Iceberg source not declared, or schema mismatch in Lakekeeper warehouse:

# Check Trino can see Iceberg tables (via Lakekeeper REST)
curl -X POST http://100.126.172.96:8091/v1/statement \
  -H "X-Trino-User: user" \
  -d "SHOW TABLES IN gravitino_iceberg.dev_streaming"

# If empty, verify Lakekeeper has the namespace:
curl -fsS 'http://192.168.100.31:8181/catalog/v1/homelab/namespaces'

"Materialized view already exists"¶

RisingWave MV already created. dbt-risingwave should handle this idempotently. If error persists, drop manually:

psql -h 100.126.172.96 -p 4566 -U root -d dev -c \
  "DROP MATERIALIZED VIEW IF EXISTS mv_name CASCADE"

Then re-run dbt run.

"Namespace does not exist" (Trino/Iceberg)¶

Trino schema doesn't exist in Lakekeeper warehouse homelab. Create it:

# Create namespace via Lakekeeper REST API
curl -fsS -X POST 'http://192.168.100.31:8181/catalog/v1/homelab/namespaces' \
  -H 'Content-Type: application/json' \
  -d '{"namespace": ["analytics"]}'

# OR create via Trino (propagates to Lakekeeper)
curl -X POST http://100.126.172.96:8091/v1/statement \
  -H "X-Trino-User: user" \
  -d "CREATE SCHEMA IF NOT EXISTS gravitino_iceberg.analytics"

dbt-risingwave specific issues¶

dbt-trino specific issues¶

Verify end-to-end after dbt changes¶

# Full stack check
scripts/dp-check.sh

# Check RW MVs
psql -h 100.126.172.96 -p 4566 -U root -d dev -c \
  "SELECT name FROM rw_catalog.rw_materialized_views ORDER BY name"

# Check Trino models
curl -X POST http://100.126.172.96:8091/v1/statement \
  -H "X-Trino-User: user" \
  -d "SHOW TABLES IN gravitino_iceberg.analytics"

Important notes¶

Sources vs Models¶

• Sources (sources.yml): Tables that already exist (Kafka sources in RW, Iceberg tables in Lakekeeper). dbt doesn't create these.
• Staging models (models/staging/): Clean and normalize raw sources. Always consume staging from downstream MVs (see "Staging layer" section above).
• Models (models/mvs/, models/marts/): New views/tables that dbt creates and manages.

RW single_node persistence¶

RisingWave runs in single_node persistent mode (v3.0.0, 8GB RAM, persisted to /mnt/user/appdata/risingwave/). Sources/MVs/sinks survive restarts. No need to re-bootstrap after restart.

Sinks not managed by dbt (mostly)¶

Iceberg sinks (RW → Iceberg via Lakekeeper) have complex WITH clauses (REST catalog URI, S3 credentials, warehouse=homelab). Most sinks stay in risingwave-init.sql, NOT in dbt. The one exception is analytics/streaming/models/sinks/iceberg_logs_sink.sql (dbt-managed). When adding a new sink in dbt, ensure the catalog.uri points at Lakekeeper :8181, NOT Gravitino :8090.

Trino federation (future)¶

Trino can federate to RisingWave via PostgreSQL connector. Add a catalog file:

# /etc/trino/catalog/risingwave.properties
connector.name=postgresql
connection-url=jdbc:postgresql://dp-risingwave:4566/dev
connection-user=root

Then dbt-trino models can SELECT FROM risingwave.public.mv_name.

Version compatibility¶