PulseStream — Real-Time Streaming Sentiment Analysis Pipeline

Ingests product reviews and social comments, scores sentiment, enriches and summarizes them with an LLM, and surfaces everything in a Streamlit dashboard.

Built in three shippable versions:

Version	What it adds
v1	Mock data → Kinesis → VADER → S3 → Snowpipe → REVIEWS_RAW → dashboard (gauge + trend)
v2	Enrichment (language, keywords, dedup) + Snowpark aggregations + Groq LLM batch summaries + insight/keyword views
v3	Real Reddit/Twitter connectors + multi-source schema + anomaly detection + comparison & alert views

producer → [Kinesis] → consumer (VADER + enrichment) → [S3 micro-batches]
   → [Snowpipe] → REVIEWS_RAW
        ├── snowpark_jobs → SENTIMENT_AGG / KEYWORD_FREQ
        ├── llm_summarizer (Groq) → INSIGHTS_SUMMARY
        └── anomaly_detection → SENTIMENT_ALERTS
   → streamlit_app (queries Snowflake)

Layout

config.py            shared env-driven config (all versions)
producer.py          v1  mock generator + publisher
consumer.py          v1  Kinesis reader + VADER + S3 writer (v2 enrichment hook)
enrichment.py        v2  langdetect + KeyBERT + dedup
llm_summarizer.py    v2  Groq batch summaries → INSIGHTS_SUMMARY
snowpark_jobs.py     v2  hourly rollup + keyword frequency
anomaly_detection.py v3  negative-sentiment spike alerts
connectors/reddit.py v3  PRAW comment stream
connectors/twitter.py v3 Tweepy filtered stream
streamlit_app.py     dashboard (all versions, degrades gracefully)
sql/snowpipe_setup.sql  v1 DDL + stage + pipe
sql/schema_v2.sql       v2 tables + JOB_STATE + optional Task
sql/schema_v3.sql       v3 schema extensions + alerts
sql/roles.sql           least-privilege roles + pipe monitoring queries
tests/                  offline pytest suite (unit + local-mode integration)
deployment_guide.md     Streamlit-in-Snowflake deployment

Setup

python3.11 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env          # then fill in credentials

On Windows: .venv\Scripts\activate, and set variables in .env rather than using the VAR=x python ... prefix shown below (which is bash-only).

Quick start — no cloud needed (local mode)

You can exercise the generate → score → micro-batch path entirely offline:

# terminal 1
LOCAL_MODE=true python consumer.py
# terminal 2
LOCAL_MODE=true python producer.py --rate 10 --count 500

Enriched micro-batch JSON files land in ./.local_sink/. To also exercise v2 enrichment locally, add ENRICHMENT_ENABLED=true (installs langdetect/KeyBERT).

v1 — full pipeline

1. Create a Kinesis stream named reviews-stream and an S3 bucket; set both in .env.
2. In Snowflake, run sql/snowpipe_setup.sql (edit the bucket/ARN placeholders), then follow its comments to wire the S3→SQS event notification using the notification_channel from DESC PIPE REVIEWS_PIPE;.
3. Start the pipeline:

   python consumer.py          # reads Kinesis, writes S3 micro-batches
   python producer.py          # publishes mock reviews
   streamlit run streamlit_app.py

v2 — enrichment, aggregation, LLM

# in Snowflake
sql/schema_v2.sql

# enable enrichment in the consumer
ENRICHMENT_ENABLED=true python consumer.py

# scheduled jobs (cron or --loop)
python snowpark_jobs.py                 # SENTIMENT_AGG + KEYWORD_FREQ
GROQ_API_KEY=... python llm_summarizer.py --loop   # INSIGHTS_SUMMARY

The dashboard's "AI insights" and keyword-cloud sections light up automatically.

v3 — real sources, anomalies

# in Snowflake
sql/schema_v3.sql

# real ingestion (each publishes to the same stream the consumer reads)
python -m connectors.reddit
python -m connectors.twitter

# anomaly alerts (depends on SENTIMENT_AGG being populated by snowpark_jobs)
python anomaly_detection.py --loop

The dashboard's comparison and alert panels then populate. For hosting the dashboard inside Snowflake, see deployment_guide.md.

Notes & design choices

• Batch LLM calls (50–100 reviews every 10 min) instead of per-event — ~98% fewer API calls, richer context, no rate-limit gymnastics.
• Snowpipe decouples the consumer from the warehouse: if Snowflake is down, records pile up safely in S3 and load when it returns.
• VADER for v1 sentiment — zero cost, in-process. Swappable for a HuggingFace model later without adding per-event API cost.
• Single-process consumer is fine for the MVP; use the Kinesis Client Library or a Lambda event-source mapping for production throughput.
• The mock producer already emits v3 fields (source_type, platform, hashed author_id, geo_region) so the schema never needs a breaking change.

Tests

pip install -r requirements-dev.txt
pytest -q            # 152 tests, no cloud accounts needed
ruff check .

The suite is deliberately offline: it covers the pure functions (sentiment thresholds, dedup hashing, record schema, LLM output validation, anomaly statistics) plus a full LOCAL_MODE integration test that runs producer → stream → consumer → sink with zero mocks. CI runs both on every push (.github/workflows/ci.yml).

Operational notes

• Mock-data + dedup footgun — with ENRICHMENT_ENABLED=true and the default DEDUP_ENABLED=true, the mock producer's small phrase bank means almost every record after the first dozen is dropped as a duplicate. Set DEDUP_ENABLED=false when demoing v2 with mock data. The consumer logs a warning when it detects this combination.
• Start the consumer before the producer on a first run — with no checkpoint yet it reads from LATEST. After that it resumes from its persisted per-shard sequence numbers.
• Rejected records are appended to rejects.jsonl rather than dropped; the consumer logs a running reject count. Malformed JSON and schema violations no longer stop ingestion.
• Set AUTHOR_HASH_KEY in .env — without it, author pseudonymization falls back to an unsalted hash that's reversible for known usernames.
• Run sql/roles.sql and switch off the default SYSADMIN role before running anything beyond a personal sandbox.
• LLM output is untrusted input — review text, especially from the v3 connectors, flows into the Groq prompt. The prompt delimits and hardens against injection and the response is schema-validated, but summaries and suggested actions are still AI-generated and unverified; the dashboard labels them as such.

Known limitations

Full audit with priorities in GAPS.md. The main open items:

• No dashboard authentication — self-hosted streamlit run exposes all sentiment data (and the Snowflake credential it holds) to anyone who can reach the port. Use Streamlit-in-Snowflake or a reverse proxy with auth (GAPS.md S2).
• Single-process consumer — it now checkpoints, retries, and survives bad records, but throughput is still bounded by one process and delivery is at-least-once. Use KCL or a Lambda event-source mapping for production volumes (GAPS.md A1).
• No containerization or process supervision — steady state is several long-running Python processes with no restart policy (GAPS.md O1).
• No retention policy or incremental aggregation — every rollup rescans REVIEWS_RAW (GAPS.md D2/D3).

What's verified vs. what needs your accounts

Verified: the test suite and lint pass, and the LOCAL_MODE path (producer → consumer → local sink) has been run end to end — including malformed and schema-violating records being dead-lettered while valid records land scored in the sink.

Not verified: anything requiring a live account. The Snowflake DDL, Snowpark jobs (including the atomic-swap refresh), Snowpipe ingestion, Groq summarization, and the Reddit/Twitter connectors have not been executed against real services. Start with local mode, then layer in the cloud services per version and validate each one.