Creating Custom Streaming Processes

Overview

Streaming processes consume messages from Kafka, transform them, and publish results back to Kafka. They bridge the consumer and producer patterns to create data transformation pipelines.

What streaming processes do:

• Subscribe to input topics
• Transform or enrich data
• Publish results to output topics
• Enable chained processing pipelines

Key difference from consumers: Streaming processes always produce output back to Kafka.

Streaming Process Architecture

Base Class Structure

All streaming processes extend KafkaStreamingProcessBase:

from smocs.cores import KafkaStreamingProcessBase

class MyStreamingProcess(KafkaStreamingProcessBase):
    def __init__(self):
        kafka_broker_url = os.getenv('KAFKA_BROKER_URL', 'kafka-broker:9092')
        group_id = 'my-streaming-group'
        input_topics = ['input-topic']
        
        super().__init__(kafka_broker_url, group_id, input_topics)
        
        # Your initialization
    
    def process_message(self, message, topic, partition, offset):
        # Your transformation logic
        # Return (success, [(output_topic, output_message)])
        pass

What's provided by base class:

• Kafka consumer setup
• Kafka producer setup (via composition)
• Message polling and publishing loop
• Input/output message validation
• Error handling and retries
• Resource cleanup

What you implement:

• process_message() - transform input to output
• State management (optional)
• External service connections (optional)

Example: Temperature Unit Converter

Scenario

Convert temperature readings from Celsius to Fahrenheit, enriching the data stream for downstream consumers.

Input message:

{
  "timestamp": 1704067200.0,
  "channels": {
    "temp_c": 22.5,
    "humidity": 45.0,
    "sensor_id": "temp-01"
  },
  "source_topic": "sensor-raw"
}

Output message:

{
  "timestamp": 1704067200.0,
  "channels": {
    "temp_c": 22.5,
    "temp_f": 72.5,
    "humidity": 45.0,
    "sensor_id": "temp-01"
  },
  "source_topic": "sensor-converted"
}

Step 1: Create Streaming Process File

Create smocs/streaming_processes/temperature_converter.py:

import os
import json
import logging
import time

from smocs.cores import KafkaStreamingProcessBase
from smocs.utils import setup_logging


class TemperatureConverterProcess(KafkaStreamingProcessBase):
    """
    Streaming process that converts Celsius to Fahrenheit.
    """
    
    def __init__(self):
        # Kafka configuration
        kafka_broker_url = os.getenv('KAFKA_BROKER_URL', 'kafka-broker:9092')
        group_id = os.getenv('CONSUMER_GROUP_ID', 'temperature-converter')
        input_topics = os.getenv('INPUT_TOPICS', 'sensor-raw').split(',')
        
        super().__init__(kafka_broker_url, group_id, input_topics)
        
        # Processing configuration
        self.output_topic = os.getenv('OUTPUT_TOPIC', 'sensor-converted')
        self.temp_channel = os.getenv('TEMP_CHANNEL', 'temp_c')
        
        logging.info("Temperature Converter initialized")
        logging.info(f"  Input topics: {input_topics}")
        logging.info(f"  Output topic: {self.output_topic}")
        logging.info(f"  Temperature channel: {self.temp_channel}")
    
    def celsius_to_fahrenheit(self, celsius):
        """Convert Celsius to Fahrenheit."""
        return (celsius * 9/5) + 32
    
    def process_message(self, message, topic, partition, offset):
        """
        Convert temperature from Celsius to Fahrenheit.
        
        Args:
            message: Message value (JSON string, already validated)
            topic: Input topic name (already validated)
            partition: Kafka partition
            offset: Message offset
            
        Returns:
            Tuple[bool, List[Tuple]]:
                - bool: True if successful, False on error
                - List: [(output_topic, output_message)] for publishing
        """
        try:
            # Parse input message
            if isinstance(message, bytes):
                message = message.decode('utf-8')
            
            data = json.loads(message)
            
            # Validate structure
            if 'channels' not in data:
                logging.warning(f"No channels in message from {topic}:{offset}")
                return False, []
            
            channels = data['channels']
            
            # Check if temperature channel exists
            if self.temp_channel not in channels:
                logging.debug(f"No {self.temp_channel} in message, passing through")
                # Pass through unchanged
                output_message = json.dumps(data)
                return True, [(self.output_topic, output_message)]
            
            # Convert temperature
            try:
                celsius = float(channels[self.temp_channel])
                fahrenheit = round(self.celsius_to_fahrenheit(celsius), 2)
                
                # Add Fahrenheit value
                fahrenheit_key = self.temp_channel.replace('_c', '_f')
                channels[fahrenheit_key] = fahrenheit
                
                logging.debug(f"Converted {celsius}°C to {fahrenheit}°F")
                
            except (ValueError, TypeError) as e:
                logging.warning(f"Invalid temperature value: {e}")
                return False, []
            
            # Create output message
            output_data = {
                'timestamp': data.get('timestamp', time.time()),
                'channels': channels,
                'source_topic': self.output_topic
            }
            
            output_message = json.dumps(output_data)
            
            # Return success and output tuple
            return True, [(self.output_topic, output_message)]
            
        except json.JSONDecodeError as e:
            logging.error(f"JSON decode error at {topic}:{offset}: {e}")
            return False, []
        except Exception as e:
            logging.error(f"Error processing message: {e}")
            return False, []


def main():
    """Main entry point."""
    setup_logging()
    logging.info("Starting Temperature Converter streaming process")
    
    process = TemperatureConverterProcess()
    process.start()


if __name__ == "__main__":
    main()

Step 2: Create Dockerfile

Create orchestration/containers/temperature-converter/Dockerfile:

FROM python:3.10-slim

WORKDIR /app

# Copy requirements
COPY orchestration/containers/temperature-converter/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy SMOCS package
COPY smocs ./smocs

# Set Python path
ENV PYTHONPATH=/app

CMD ["python", "-m", "smocs.streaming_processes.temperature_converter"]

Step 3: Create Requirements File

Create orchestration/containers/temperature-converter/requirements.txt:

kafka-python==2.0.2

Step 4: Add to Docker Compose

Add to orchestration/docker-compose.yml:

temperature-converter:
  build:
    context: ../
    dockerfile: ./orchestration/containers/temperature-converter/Dockerfile
  profiles: ["converter"]
  environment:
    KAFKA_BROKER_URL: kafka-broker:9092
    CONSUMER_GROUP_ID: temperature-converter
    INPUT_TOPICS: sensor-raw
    OUTPUT_TOPIC: sensor-converted
    TEMP_CHANNEL: temp_c
    SMOCS_LOG_LEVEL: INFO
  networks:
    - kafka-net
  depends_on:
    kafka-broker:
      condition: service_healthy
  restart: unless-stopped

Step 5: Configure Environment

Add to .env:

COMPOSE_PROFILES=converter

Step 6: Test Streaming Process

# Build and start
docker compose up --build temperature-converter

# In another terminal, publish test message to input topic
docker exec kafka-broker kafka-console-producer.sh \
  --topic sensor-raw \
  --bootstrap-server localhost:9092 << EOF
{"timestamp": 1704067200.0, "channels": {"temp_c": 22.5, "humidity": 45.0}}
EOF

# Consume from output topic to verify transformation
docker exec kafka-broker kafka-console-consumer.sh \
  --topic sensor-converted \
  --bootstrap-server localhost:9092 \
  --from-beginning \
  --max-messages 1

# Expected output includes both temp_c and temp_f
# {"timestamp": 1704067200.0, "channels": {"temp_c": 22.5, "temp_f": 72.5, "humidity": 45.0}, ...}

# Check logs for processing confirmation
docker compose logs temperature-converter | grep "Converted"

Key Implementation Details

Return Format

The process_message() method must return a tuple:

return (success_bool, list_of_output_tuples)

Success boolean: True if processing succeeded, False on error

Output tuples: Each tuple is (output_topic, output_message) or (output_topic, output_message, key)

Examples:

# Single output
return True, [('output-topic', json.dumps(data))]

# Multiple outputs to different topics
return True, [
    ('topic-1', json.dumps(data1)),
    ('topic-2', json.dumps(data2))
]

# With message key
return True, [('output-topic', json.dumps(data), 'sensor-01')]

# No output (filtering)
return True, []

# Processing failed
return False, []

Message Validation

Both input and output messages are automatically validated by the base class:

• Input validation occurs before process_message() is called
• Output validation occurs before publishing to Kafka
• Invalid messages are logged and skipped

Error Handling

Processing errors: Return False, [] to indicate failure

Pass-through on error: Return the original message unchanged

try:
    # Process message
    result = self.transform(data)
    return True, [(self.output_topic, json.dumps(result))]
except Exception as e:
    logging.warning(f"Transform failed: {e}, passing through")
    return True, [(self.output_topic, message)]

Common Patterns

Conditional Output

def process_message(self, message, topic, partition, offset):
    data = json.loads(message)
    
    # Only output if condition met
    if data['channels'].get('temperature', 0) > 30:
        return True, [(self.output_topic, json.dumps(data))]
    
    # Filter out (no output)
    return True, []

Multi-Topic Output

def process_message(self, message, topic, partition, offset):
    data = json.loads(message)
    
    # Send to different topics based on content
    outputs = []
    
    if data['channels'].get('is_alert'):
        outputs.append(('alerts', json.dumps(data)))
    else:
        outputs.append(('normal-data', json.dumps(data)))
    
    return True, outputs

Stateful Processing

def __init__(self):
    super().__init__(...)
    self.message_count = 0

def process_message(self, message, topic, partition, offset):
    self.message_count += 1
    
    data = json.loads(message)
    data['channels']['message_number'] = self.message_count
    
    return True, [(self.output_topic, json.dumps(data))]

Best Practices

Keep transformations fast: Process messages in milliseconds

Preserve SMOCS format: Maintain timestamp, channels, source_topic structure

Handle missing fields: Don't fail on missing data, use defaults

Log processing metrics: Track throughput and errors

Idempotent operations: Handle duplicate messages gracefully

Return empty list to filter: Use return True, [] to skip output

Use explicit topic names: Avoid dynamic topic creation when possible