# anuj690/customer-behavior-platform architecture

> Architecture diagram of the GitHub repository anuj690/customer-behavior-platform, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-09-25). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/anuj690/customer-behavior-platform
- Repository: https://github.com/anuj690/customer-behavior-platform

## Overview

The Customer Behavior Platform generates customer journey events, publishes them to Kafka, and processes them with Spark Structured Streaming into lakehouse layers and analytical outputs. The main path is producer generation and publication, stream parsing and validation, deduplication, then Bronze/Silver/Gold and analytics writes; invalid records go to a dead-letter store. The graph follows the documented architecture and sampled orchestration. Some streaming-service wiring is omitted from the excerpt, so uncertain internal calls are not shown.

Main entry points: [`streaming-service/src/streaming_service/service.py`](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/service.py) (Batch orchestrator), [`streaming-service/src/streaming_service/iceberg/tables.py`](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/iceberg/tables.py) (Iceberg tables), [`producer-service/src/producer_service/service.py`](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/service.py) (Journey publisher), [`streaming-service/src/streaming_service/app.py`](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/app.py) (Streaming entry)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_generation["Event Generation"]
  node_producer_app["Producer entry<br/>[app.py]"]
  node_producer_service["Journey publisher<br/>[service.py]"]
  node_customer_gen["Customer generator<br/>[customer.py]"]
  node_session_gen["Session generator<br/>[session.py]"]
  node_journey_gen["Journey generator<br/>[journey.py]"]
  node_event_factory["Event factory<br/>[event_factory.py]"]
  node_corruptor["Event corruptor<br/>[event_corruptor.py]"]
  node_kafka_producer["Kafka publisher<br/>[producer.py]"]
end

subgraph group_ingestion["Stream Ingestion"]
  node_stream_app["Streaming entry<br/>[app.py]"]
  node_stream_service["Batch orchestrator<br/>[service.py]"]
  node_consumer["Kafka consumer<br/>[consumer.py]"]
  node_parser["JSON parser<br/>[event_parser.py]"]
  node_transformer["Event transformer"]
  node_validator["Event validator<br/>[validator.py]"]
end

subgraph group_lakehouse["Lakehouse Processing"]
  node_dedup["Deduplication<br/>[writer.py]"]
  node_bronze["Bronze writer<br/>[writer.py]"]
  node_silver["Silver transform<br/>[transformer.py]"]
  node_silver_writer["Silver writer<br/>[writer.py]"]
  node_gold["Gold aggregation<br/>[aggregator.py]"]
  node_gold_writer["Gold writer<br/>[writer.py]"]
  node_dlq["Dead-letter writer<br/>[writer.py]"]
end

subgraph group_analytics["Analytics"]
  node_dimensions["Dimension outputs"]
  node_facts["Fact events<br/>[fact_events.py]"]
  node_daily_metrics["Daily metrics"]
  node_funnel_metrics["Funnel metrics<br/>[funnel_metrics.py]"]
end

subgraph group_platform["Runtime Platform"]
  node_iceberg[("Iceberg tables<br/>[tables.py]")]
  node_spark_factory["Spark session<br/>[spark_session.py]"]
end

node_user(("Event source"))
node_kafka["Kafka topic"]
node_s3[("Amazon S3")]

node_user -->|"initiates"| node_producer_app
node_producer_app -->|"starts"| node_producer_service
node_producer_service -->|"generates customer"| node_customer_gen
node_producer_service -->|"creates session"| node_session_gen
node_producer_service -->|"generates journey"| node_journey_gen
node_journey_gen -->|"builds events"| node_event_factory
node_producer_service -->|"corrupts events"| node_corruptor
node_producer_service -->|"publishes events"| node_kafka_producer
node_kafka_producer -->|"sends events"| node_kafka
node_stream_app -->|"creates session"| node_spark_factory
node_stream_app -->|"starts"| node_stream_service
node_stream_service -->|"reads stream"| node_consumer
node_consumer -->|"consumes events"| node_kafka
node_stream_service -->|"parses stream"| node_parser
node_parser -->|"supplies records"| node_transformer
node_stream_service -->|"validates records"| node_validator
node_stream_service -->|"deduplicates valid"| node_dedup
node_stream_service -->|"writes invalid"| node_dlq
node_stream_service -->|"writes batch"| node_bronze
node_stream_service -->|"transforms batch"| node_silver
node_silver -->|"writes records"| node_silver_writer
node_stream_service -->|"aggregates records"| node_gold
node_gold -->|"writes aggregates"| node_gold_writer
node_stream_service -->|"writes dimensions"| node_dimensions
node_stream_service -->|"writes facts"| node_facts
node_stream_service -->|"writes metrics"| node_daily_metrics
node_stream_service -->|"writes funnel"| node_funnel_metrics
node_bronze -->|"merges records"| node_iceberg
node_silver_writer -->|"writes layer"| node_iceberg
node_gold_writer -->|"writes layer"| node_iceberg
node_dlq -->|"stores invalid"| node_iceberg
node_dimensions -->|"writes tables"| node_iceberg
node_facts -->|"writes fact table"| node_iceberg
node_daily_metrics -->|"writes metrics"| node_iceberg
node_funnel_metrics -->|"merges metrics"| node_iceberg
node_iceberg -->|"stores data"| node_s3

click node_producer_app "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/app.py"
click node_producer_service "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/service.py"
click node_customer_gen "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/generators/customer.py"
click node_session_gen "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/generators/session.py"
click node_journey_gen "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/generators/journey.py"
click node_event_factory "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/factories/event_factory.py"
click node_corruptor "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/utils/event_corruptor.py"
click node_kafka_producer "https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/kafka/producer.py"
click node_stream_app "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/app.py"
click node_stream_service "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/service.py"
click node_consumer "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/kafka/consumer.py"
click node_parser "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/transforms/event_parser.py"
click node_transformer "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/transforms/event_transformer.py"
click node_validator "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/transforms/validator.py"
click node_dedup "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/deduplication/writer.py"
click node_bronze "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/bronze/writer.py"
click node_silver "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/silver/transformer.py"
click node_silver_writer "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/silver/writer.py"
click node_gold "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/gold/aggregator.py"
click node_gold_writer "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/gold/writer.py"
click node_dlq "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/dead_letter/writer.py"
click node_dimensions "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/dimension_writer.py"
click node_facts "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/fact_events.py"
click node_daily_metrics "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/daily_user_metrics.py"
click node_funnel_metrics "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/funnel_metrics.py"
click node_iceberg "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/iceberg/tables.py"
click node_spark_factory "https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/spark/spark_session.py"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_producer_app,node_producer_service,node_customer_gen,node_session_gen,node_journey_gen,node_event_factory,node_corruptor,node_kafka_producer toneBlue
class node_stream_app,node_stream_service,node_consumer,node_parser,node_transformer,node_validator,node_s3 toneAmber
class node_dedup,node_bronze,node_silver,node_silver_writer,node_gold,node_gold_writer,node_dlq toneMint
class node_dimensions,node_facts,node_daily_metrics,node_funnel_metrics toneRose
class node_iceberg,node_spark_factory,node_user,node_kafka toneIndigo
```

## Components

### Event Generation

- [Producer entry](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/app.py): `producer-service/src/producer_service/app.py`
- [Journey publisher](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/service.py): `producer-service/src/producer_service/service.py`
- [Customer generator](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/generators/customer.py): `producer-service/src/producer_service/generators/customer.py`
- [Session generator](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/generators/session.py): `producer-service/src/producer_service/generators/session.py`
- [Journey generator](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/generators/journey.py): `producer-service/src/producer_service/generators/journey.py`
- [Event factory](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/factories/event_factory.py): `producer-service/src/producer_service/factories/event_factory.py`
- [Event corruptor](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/utils/event_corruptor.py): `producer-service/src/producer_service/utils/event_corruptor.py`
- [Kafka publisher](https://github.com/anuj690/customer-behavior-platform/blob/main/producer-service/src/producer_service/kafka/producer.py): `producer-service/src/producer_service/kafka/producer.py`

### Stream Ingestion

- [Streaming entry](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/app.py): `streaming-service/src/streaming_service/app.py`
- [Batch orchestrator](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/service.py): `streaming-service/src/streaming_service/service.py`
- [Kafka consumer](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/kafka/consumer.py): `streaming-service/src/streaming_service/kafka/consumer.py`
- [JSON parser](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/transforms/event_parser.py): `streaming-service/src/streaming_service/transforms/event_parser.py`
- [Event transformer](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/transforms/event_transformer.py): `streaming-service/src/streaming_service/transforms/event_transformer.py`
- [Event validator](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/transforms/validator.py): `streaming-service/src/streaming_service/transforms/validator.py`

### Lakehouse Processing

- [Deduplication](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/deduplication/writer.py): `streaming-service/src/streaming_service/deduplication/writer.py`
- [Bronze writer](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/bronze/writer.py): `streaming-service/src/streaming_service/bronze/writer.py`
- [Silver transform](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/silver/transformer.py): `streaming-service/src/streaming_service/silver/transformer.py`
- [Silver writer](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/silver/writer.py): `streaming-service/src/streaming_service/silver/writer.py`
- [Gold aggregation](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/gold/aggregator.py): `streaming-service/src/streaming_service/gold/aggregator.py`
- [Gold writer](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/gold/writer.py): `streaming-service/src/streaming_service/gold/writer.py`
- [Dead-letter writer](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/dead_letter/writer.py): `streaming-service/src/streaming_service/dead_letter/writer.py`

### Analytics

- [Dimension outputs](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/dimension_writer.py): `streaming-service/src/streaming_service/analytics/dimension_writer.py`
- [Fact events](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/fact_events.py): `streaming-service/src/streaming_service/analytics/fact_events.py`
- [Daily metrics](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/daily_user_metrics.py): `streaming-service/src/streaming_service/analytics/daily_user_metrics.py`
- [Funnel metrics](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/analytics/funnel_metrics.py): `streaming-service/src/streaming_service/analytics/funnel_metrics.py`

### Runtime Platform

- [Iceberg tables](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/iceberg/tables.py): `streaming-service/src/streaming_service/iceberg/tables.py`
- [Spark session](https://github.com/anuj690/customer-behavior-platform/blob/main/streaming-service/src/streaming_service/spark/spark_session.py): `streaming-service/src/streaming_service/spark/spark_session.py`

### Other components

- **Event source**
- **Kafka topic**
- **Amazon S3**

## Connections

- Event source → Producer entry: initiates
- Producer entry → Journey publisher: starts
- Journey publisher → Customer generator: generates customer
- Journey publisher → Session generator: creates session
- Journey publisher → Journey generator: generates journey
- Journey generator → Event factory: builds events
- Journey publisher → Event corruptor: corrupts events
- Journey publisher → Kafka publisher: publishes events
- Kafka publisher → Kafka topic: sends events
- Streaming entry → Spark session: creates session
- Streaming entry → Batch orchestrator: starts
- Batch orchestrator → Kafka consumer: reads stream
- Kafka consumer → Kafka topic: consumes events
- Batch orchestrator → JSON parser: parses stream
- JSON parser → Event transformer: supplies records
- Batch orchestrator → Event validator: validates records
- Batch orchestrator → Deduplication: deduplicates valid
- Batch orchestrator → Dead-letter writer: writes invalid
- Batch orchestrator → Bronze writer: writes batch
- Batch orchestrator → Silver transform: transforms batch
- Silver transform → Silver writer: writes records
- Batch orchestrator → Gold aggregation: aggregates records
- Gold aggregation → Gold writer: writes aggregates
- Batch orchestrator → Dimension outputs: writes dimensions
- Batch orchestrator → Fact events: writes facts
- Batch orchestrator → Daily metrics: writes metrics
- Batch orchestrator → Funnel metrics: writes funnel
- Bronze writer → Iceberg tables: merges records
- Silver writer → Iceberg tables: writes layer
- Gold writer → Iceberg tables: writes layer
- Dead-letter writer → Iceberg tables: stores invalid
- Dimension outputs → Iceberg tables: writes tables
- Fact events → Iceberg tables: writes fact table
- Daily metrics → Iceberg tables: writes metrics
- Funnel metrics → Iceberg tables: merges metrics
- Iceberg tables → Amazon S3: stores data