Skip to main content

Data Ingest Thread

Core Responsibilities​

The Data Ingest Thread serves as the agent's primary interface to the streaming data ecosystem. Built on KafkaConsumerBase, it maintains a persistent connection to configured Kafka topics and transforms raw streaming messages into structured database records.

Processing Pipeline​

The thread operates through a continuous polling loop that retrieves message batches from Kafka. Each message undergoes parsing to extract sensor readings, timestamps, and metadata. The AutoencoderDataIngestThread implementation demonstrates this pattern by parsing JSON messages containing gymnasium environment state data, extracting numeric state values while filtering out non-numeric metadata fieldsI wou.

Data Storage Strategy​

Parsed data flows into MySQL through the shared DBManager instance. The thread uses the record_sensor_data() method to store timestamped sensor readings with proper data type conversion (numpy arrays to binary blobs). This creates a persistent training dataset that accumulates over time, providing the foundation for ML model development.

Error Handling and Resilience​

The thread implements robust error handling at multiple levels: JSON parsing failures are logged but don't terminate processing, database connection issues trigger retry logic, and malformed messages are skipped with appropriate warnings. This ensures that transient data quality issues don't disrupt the overall data flow.

Configuration Integration​

Topic subscriptions, parsing rules, and storage parameters are all driven by the central configuration file. This allows agents to be reconfigured for different data sources without code changes, supporting the system's flexibility goals.

User Implementation Requirements​

Single Required Method​

def store_message(self, message, topic, partition, offset) -> bool:
    # Parse message, validate data, store to database
    # Return True for success, False for failure

Implementation Steps​

  1. Parse Message: Decode bytes to string, parse JSON/format
  2. Extract Data: Pull relevant fields (timestamps, sensor values, metadata)
  3. Validate: Check data types, handle missing fields, filter invalid data
  4. Transform: Convert to database schema format (arrays to numpy, timestamps to datetime)
  5. Store: Call self.db_manager.record_sensor_data(data_dict)
  6. Return Status: True if successful, False if failed

Configuration Needed​

config = {
    'kafka_topics': {'input': 'your-topic-name'}
}

Environment Variables Required​

  • KAFKA_BROKER_URL
  • MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_ROOT_PASSWORD, MYSQL_DATABASE

What's Handled Automatically​

  • Kafka consumer setup/teardown
  • Topic subscription and polling
  • Database connection management
  • Thread lifecycle and health monitoring
  • Error recovery and restart logic
  • Message batching and offset management

Key Constraints​

  • Single-threaded: Keep store_message() fast and efficient
  • Error handling: Catch exceptions, log errors, return False for failures
  • Database schema: Match expected format for record_sensor_data()
  • Memory management: Don't accumulate state between messages

The base classes handle all infrastructure complexity - users only implement domain-specific data transformation logic.

Architectual Loop Diagram​

SMOCS Data Ingest Thread Diagram