Sientia DataOps Scouter

The Sientia DataOps Scouter is a Temporal-based workflow application that processes industrial data from OPC collectors. It aggregates and filters data received from OPC collectors via Kafka topics, direct access to the OPC server, or active trigger (real time applications). The processed data is then stored or forwarded for further analysis.

Key Features

• Data ingestion from multiple sources:
  • OPC collectors through Kafka
  • Direct access to OPC servers
  • Real-time triggers for immediate processing
• Data aggregation and filtering
• Workflow orchestration using Temporal.io
• Integration with Redis for caching and PostgreSQL for storage
• Scalable deployment using Kubernetes

Workflows

Core Scouter

The Core Scouter is the main workflow processes the received data. Steps:

• data_quality_gate: Filters the data received from the OPC collector.
• aggregate_data: Aggregates the data received from the OPC collector.
• group_and_hold_data: Groups the data received from the OPC collector.
• export_data_to_postgres: Exports the data received from the OPC collector to PostgreSQL.

Scouter

The Scouter is the batch basic workflow that extracts data from the source and processes it using the Core Scouter workflow. Steps:

• load_from_kafka: Loads data from a kafka topic.
• core_scouter: Processes the data using the Core Scouter workflow.

Workflow inputs:

• topic (str): Kafka topic name where data is received
• schedule_name (str): Name of the schedule that triggers the workflow
• model_name (str): Name of the model being used for processing
• model_id (int): Unique identifier for the model
• trigger_laborious (bool): Flag indicating if laborious direct processing is required (real time applications)
• filters (dict): Dictionary containing data filtering rules
  • NULL_VALUES_FILTER: Configuration for handling null values
    • policy: Policy for null values ("KEEP" or "DISCARD")
  • OUT_OF_BOUNDS_FILTER: Configuration for handling out-of-bounds values
    • policy: Policy for out-of-bounds values ("KEEP" or "DISCARD")
• schema (str): Database schema name where data will be stored
• table_name (str): Name of the table where data will be stored
• retention_time (int): Time in seconds that data will be retained in Redis
• model_tags (dict): Configuration for different OPC tags
  • The key is the tag name and the value is a dictionary containing:
    • data_range: List of two numbers [min, max] defining valid data range
    • aggr_function: Aggregation function to use ("lts", "mdn", "avg", "max", "min")

Fake Data

The Fake Data activity is used to generate fake data for testing purposes. Steps:

• generate_and_send_data: Generates fake data and sends it to a kafka topic.

Workflow inputs:

• topic (str): Kafka topic name where data is received

Environment variables

• POSTGRES_HOST

• POSTGRES_PORT

• POSTGRES_USER

• POSTGRES_PASSWORD

• POSTGRES_DBNAME

• POSTGRES_MIN_CONNECTIONS

• POSTGRES_MAX_CONNECTIONS

• KAFKA_BOOTSTRAP_SERVERS

• KAFKA_POLLING_TIME

• REDIS_HOST

• REDIS_PORT

• REDIS_USERNAME

• REDIS_PASSWORD

• LOG_LEVEL

• PROJECT_NAME

• TEMPORAL_HOST

• TEMPORAL_NAMESPACE

Application deployment

The application can be deployed using the following command:

helm upgrade --install sientia-dataops-opc-ingestor sientia/sientia-module -n sientia-opc --create-namespace -f ./values.yaml