How to Ingest Data from Streaming Sources Like Kafka with Semantica

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 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) 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:

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:

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, allowing dynamic source selection. Unit tests in 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.
• 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. 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}