# How to Ingest Data from Streaming Sources Like Kafka with Semantica

> Learn how to ingest data from streaming sources like Kafka with Semantica's unified framework. Explore StreamIngestor and StreamMonitor for efficient data pipelines.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: how-to-guide
- Published: 2026-08-22

---

**Semantica provides a unified streaming-ingestion framework built around the `StreamIngestor` class, which orchestrates source-specific processors and monitors pipeline health through a centralized `StreamMonitor`.**

Semantica is an open-source AI framework designed to simplify real-time data ingestion. This guide explains how to ingest data from streaming sources like Kafka using Semantica's pluggable architecture, which abstracts consumer management while maintaining full control over transformation, validation, and error handling.

## Architecture of the Streaming Framework

Semantica's streaming implementation resides in [`semantica/ingest/stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/stream_ingestor.py) and consists of three core components that work together to provide reliable, observable data ingestion.

### StreamIngestor Orchestration

The **`StreamIngestor`** class serves as the central orchestrator. According to the source code at `semantica-agi/semantica`, the constructor at lines 41-56 initializes a `StreamMonitor` instance that tracks all registered processors. The ingestor exposes the `ingest_kafka()` method (defined at lines 64-78), which dynamically instantiates a `KafkaProcessor` and registers it with the monitor.

### KafkaProcessor Implementation

The **`KafkaProcessor`** class extends `StreamProcessor` and wraps the official `kafka-python` `KafkaConsumer`. The constructor (lines 205-240 in [`stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/stream_ingestor.py)) builds the consumer and stores configuration, while the `_consume_loop()` method (lines 234-247) polls messages continuously. Each message passes through `StreamProcessor.process_message()` before forwarding to your handler, ensuring consistent validation and transformation across all streaming sources.

### StreamMonitor Health Tracking

The **`StreamMonitor`** aggregates statistics from all active processors. When you register a processor via `monitor_processor()` (lines 29-41), the monitor begins tracking processed counts, error rates, and overall health status. You can query pipeline health through `ingestor.monitor.check_health()`, which evaluates error rates against configurable thresholds to determine if streams are healthy or degraded.

## Step-by-Step Kafka Ingestion

Setting up a Kafka ingestion pipeline requires five specific steps:

1. **Instantiate `StreamIngestor`** – Creates the orchestrator and monitoring infrastructure.
2. **Create a Kafka processor** – Call `ingest_kafka(topic, bootstrap_servers, **options)` to configure connection parameters.
3. **Set a message handler** – Use `processor.set_message_handler()` to define a callback that receives parsed payloads.
4. **Start streaming** – Invoke `ingestor.start_streaming()` to launch background threads for each processor.
5. **Monitor health** – Periodically check `ingestor.monitor.check_health()` to ensure error rates remain within acceptable bounds.

## Practical Implementation

The following example demonstrates a complete Kafka ingestion setup using the `StreamIngestor` API:

```python
from semantica.ingest import StreamIngestor

# 1️⃣ Initialise the ingestor

ingestor = StreamIngestor()

# 2️⃣ Create a Kafka processor for topic "events"

kafka_proc = ingestor.ingest_kafka(
    topic="events",
    bootstrap_servers=["localhost:9092"],
    # optional: custom deserialization or consumer config

    consumer_config={"auto_offset_reset": "earliest"},
)

# 3️⃣ Define a simple message handler

def handle_message(msg: dict) -> None:
    # msg contains {"content": ..., "metadata": ..., "processed_at": ...}

    print("Received:", msg["content"])

kafka_proc.set_message_handler(handle_message)

# 4️⃣ Start the streaming pipeline (runs in background threads)

ingestor.start_streaming()

# 5️⃣ Periodically check health (e.g., from a separate thread or scheduler)

health = ingestor.monitor.check_health()
print("Stream health:", health["overall"])

```

## Advanced Data Processing

Semantica supports inline transformation and validation through the processor configuration. These functions execute within the consumption loop before your message handler receives the data:

```python
def to_uppercase(data):
    data["payload"] = data["payload"].upper()
    return data

def validate(data):
    return "payload" in data and isinstance(data["payload"], str)

kafka_proc = ingestor.ingest_kafka(
    "events",
    ["localhost:9092"],
    transform=to_uppercase,
    validate=validate,
)
kafka_proc.set_message_handler(handle_message)
ingestor.start_streaming()

```

## Monitoring Pipeline Health

The `StreamMonitor` provides real-time visibility into pipeline performance. Each processor reports statistics to the monitor, which aggregates them into health metrics. The `check_health()` method returns an overall status based on error rate thresholds, enabling automatic alerting when streams degrade. This centralized approach eliminates the need to instrument each consumer individually.

The framework also registers the `"kafka"` method with the generic `ingest_stream` entry point in [`semantica/ingest/methods.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/methods.py), allowing dynamic source selection. Unit tests in [`tests/ingest/test_submodules.py`](https://github.com/semantica-agi/semantica/blob/main/tests/ingest/test_submodules.py) demonstrate mock-based verification of the `ingest_kafka` workflow.

## Summary

- **Unified Architecture**: The `StreamIngestor` class orchestrates multiple streaming sources through a common interface defined in [`semantica/ingest/stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/stream_ingestor.py).
- **Kafka Integration**: The `KafkaProcessor` wraps `kafka-python` consumers and implements `_consume_loop()` for continuous message processing.
- **Pluggable Processing**: Transform and validate functions execute automatically before messages reach your handler.
- **Centralized Monitoring**: `StreamMonitor` tracks health metrics across all processors via `monitor_processor()` and exposes health status through `check_health()`.
- **Extensible Design**: Adding new streaming sources requires only a new `*Processor` subclass and registration in `StreamIngestor`.

## Frequently Asked Questions

### How does Semantica handle Kafka consumer failures?

The `KafkaProcessor` runs inside a managed thread that catches exceptions during the `_consume_loop()` and reports them to the `StreamMonitor`. The monitor tracks error rates and marks the stream as unhealthy when thresholds are exceeded, allowing you to implement circuit breakers or alerting logic based on `ingestor.monitor.check_health()` results.

### Can I ingest from multiple Kafka topics simultaneously?

Yes. Instantiate multiple processors by calling `ingestor.ingest_kafka()` with different topic names or connection parameters. Each processor registers independently with the `StreamMonitor`, and `start_streaming()` launches a separate background thread for each one. The monitor aggregates health metrics across all active streams.

### What message format does the handler callback receive?

The `set_message_handler()` callback receives a dictionary containing three keys: `"content"` (the parsed payload), `"metadata"` (Kafka headers and partition info), and `"processed_at"` (timestamp). This standardized format applies across all streaming sources in Semantica, ensuring handler code remains portable between Kafka, Pulsar, RabbitMQ, and Kinesis.

### Where is the Kafka ingestion logic tested?

The repository includes unit tests for streaming ingestion in [`tests/ingest/test_submodules.py`](https://github.com/semantica-agi/semantica/blob/main/tests/ingest/test_submodules.py). These tests verify the `ingest_kafka` method and mock the `KafkaProcessor` to validate integration with the `StreamIngestor` orchestration layer without requiring a live Kafka cluster.