# How to Ingest Data from Kafka Streams Using Semantica: A Complete Guide

> Learn how to ingest data from Kafka streams using Semantica. This guide covers using StreamIngestor and KafkaProcessor for efficient data consumption with health monitoring.

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

---

**To ingest data from Kafka streams using Semantica, instantiate the `StreamIngestor` class and call `ingest_kafka()`, or use the `ingest_stream` convenience method to create a `KafkaProcessor`, register a message handler with `set_message_handler()`, and invoke `start_consuming()` to begin polling with built-in health monitoring.**

Semantica provides a flexible, extensible streaming ingestion framework for building real-time AI applications. This guide demonstrates the complete workflow to ingest data from Kafka streams using Semantica's Python API, referencing the actual implementation in the `semantica-agi/semantica` repository.

## Core Architecture Components

### StreamIngestor

The `StreamIngestor` class defined in [`semantica/ingest/stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/stream_ingestor.py) (lines 19-38) serves as the primary orchestrator for stream processing. It creates and registers stream processors and attaches a health-monitoring layer that tracks runtime metrics across all active connections.

### KafkaProcessor

Defined in [`semantica/ingest/stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/stream_ingestor.py) (lines 25-48), `KafkaProcessor` is a concrete subclass of `StreamProcessor` that wraps a `kafka.KafkaConsumer`. It polls records from subscribed topics, converts them to the internal `StreamMessage` representation, applies optional transformations and validations, and forwards processed payloads to a user-provided handler function.

### StreamMonitor

The `StreamMonitor` component maintains runtime metrics including processed message count, error rate, and health status for every active processor (lines 29-41). It enables integration with external observability tools by exposing structured health data through the `check_health()` method.

## Ingestion Methods

Semantica offers two primary approaches to initialize Kafka ingestion. Choose based on your control requirements:

### High-Level: ingest_stream Convenience Method

**`ingest_stream`** – Use this function from `semantica.ingest.methods` for rapid prototyping and simple configurations. This wrapper internally creates a `StreamIngestor` and instantiates the appropriate processor based on the `method` argument.

```python
from semantica.ingest.methods import ingest_stream

# Define the Kafka source configuration

kafka_source = {
    "topic": "my_topic",
    "bootstrap_servers": ["localhost:9092"]
}

# Create a processor (KafkaProcessor under the hood)

processor = ingest_stream(kafka_source, method="kafka")

# Register a handler that receives the normalized message payload

def handle_message(msg: dict) -> None:
    print("Received:", msg["content"])

processor.set_message_handler(handle_message)

# Start the consumer loop

processor.start_consuming()

```

According to [`semantica/ingest/methods.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/methods.py) (lines 64-76), this wrapper handles the boilerplate of processor selection while exposing the same configuration options as the direct API.

### Direct API: StreamIngestor Class

**`StreamIngestor`** – Use this class directly when managing multiple processors simultaneously or when you need to inject global configuration at the ingestor level. The `ingest_kafka()` method returns a configured `KafkaProcessor` instance that you can customize before starting consumption.

```python
from semantica.ingest.stream_ingestor import StreamIngestor

# Initialise the ingestor (can accept global config here)

ingestor = StreamIngestor()

# Obtain a Kafka processor for the desired topic

kafka_proc = ingestor.ingest_kafka(
    topic="my_topic",
    bootstrap_servers=["localhost:9092"],
    # Optional: custom validation or transformation functions

    transform=lambda d: {"value": d["value"] * 2},
    validate=lambda d: isinstance(d.get("value"), int)
)

# Define a handler to process each message after validation/transformation

def process(msg: dict) -> None:
    print("Processed:", msg)

kafka_proc.set_message_handler(process)

# Start the ingestion loop (multiple processors can be started together)

ingestor.start_streaming()

```

The `StreamIngestor.__init__` method in [`semantica/ingest/stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/stream_ingestor.py) (lines 41-55) accepts global configuration, while `ingest_kafka` (lines 64-78) supports optional `transform` and `validate` callables that execute sequentially on each message before reaching your handler. The `start_streaming()` method (lines 38-49) initiates consumption for all registered processors in the ingestor's pool.

## Monitoring Stream Health

After initiating ingestion, query the `StreamMonitor` to retrieve real-time metrics and health status. This is essential for production deployments requiring observability integration.

```python

# After starting the ingestion as in the Direct API example

import time

while True:
    health = ingestor.monitor.check_health()
    print("Overall status:", health["overall"])
    for name, stats in health["processors"].items():
        print(f"{name}: running={stats['running']}, errors={stats['errors']}")
    time.sleep(30)   # poll every 30 seconds

```

The `check_health()` method in [`semantica/ingest/stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/stream_ingestor.py) (lines 73-86) returns a dictionary containing an `overall` status string and a `processors` dictionary with per-processor statistics including `running` boolean flags and cumulative `errors` counts.

## Summary

- The `StreamIngestor` class in [`semantica/ingest/stream_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/stream_ingestor.py) serves as the central orchestrator for Kafka ingestion workflows.
- Use **`ingest_stream`** from `semantica.ingest.methods` for quick setup, or instantiate **`StreamIngestor`** directly for fine-grained control over multiple processors and global configuration.
- The **`KafkaProcessor`** wraps the underlying Kafka client, converts records to `StreamMessage` objects, and supports optional `transform` and `validate` functions.
- Access runtime health metrics and error rates through **`StreamMonitor.check_health()`** to integrate with external monitoring systems.

## Frequently Asked Questions

### How do I start consuming messages from a Kafka topic in Semantica?

Create a processor using either `ingest_stream()` or `StreamIngestor().ingest_kafka()`, register your processing logic via `set_message_handler()`, and call `start_consuming()` on the processor or `start_streaming()` on the ingestor. The framework handles polling, message conversion, and error handling automatically.

### What is the difference between ingest_stream and StreamIngestor?

**`ingest_stream`** is a convenience function in [`semantica/ingest/methods.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/methods.py) that instantiates a `StreamIngestor` internally and returns a ready-to-use processor. **`StreamIngestor`** is the underlying class that manages multiple processors and global state; use it directly when you need to coordinate several Kafka topics or share configuration across processors.

### Can I apply custom transformations to Kafka messages before processing?

Yes. Both the `ingest_kafka()` method and the high-level `ingest_stream()` function accept optional `transform` and `validate` parameters. The `transform` function receives the raw message dictionary and must return a modified dictionary, while `validate` returns a boolean determining whether the message should proceed to your handler.

### How does Semantica handle health monitoring for Kafka streams?

Semantica attaches a `StreamMonitor` instance to every `StreamIngestor` that tracks processed counts, error rates, and running status for each processor. Call `ingestor.monitor.check_health()` to retrieve a structured health report suitable for integration with Prometheus, Datadog, or other observability platforms.