Step 9: Streaming feature pipeline (Pub/Sub → Feature Store)

[shlbatra]

Summary

• Add src/dataflow/iris_feature_pipeline.py — Beam streaming pipeline that reads from Pub/Sub, validates messages with Pydantic (dataflow/models/iris_schema.py), renames fields to canonical names, and dual-writes to:
  • BQ (offline store) — WRITE_APPEND for training/batch
  • Bigtable (online store) — direct WriteFeatureValues API for sub-second real-time serving
• Add src/dataflow/models/iris_schema.py — PubSubIrisMessage Pydantic model for type-safe Pub/Sub message validation
• Add src/dataflow/utils/online_store_writer.py — reusable WriteToOnlineStore Beam DoFn for direct Bigtable writes
• Add .github/workflows/deploy-dataflow-feature.yaml — GitHub Action (workflow_dispatch) to deploy the feature pipeline
• Add scripts/deploy_dataflow_feature.sh — local deploy script

Key design decisions

• Dual-write instead of periodic sync — writes to BQ and Bigtable in parallel, no 3–10 min sync delay
• source = "streaming" distinguishes streaming-ingested rows from batch ("training", "batch_input")
• entity_id = "{sample_id}_streaming" — uses Pub/Sub message sample_id when available, falls back to UUID
• BatchElements groups rows for efficient online store writes without adding latency at low throughput
• No model calls — this pipeline only persists features; inference is handled separately

Test plan

• Deploy to staging via GitHub Action or local script
• Publish messages with pubsub_producer.py
• Verify rows appear in iris_features BQ table with source = 'streaming'
• Verify features are available in online store for real-time lookups

🤖 Generated with Claude Code