4352c67f3a
Enhance PI Web API Scouter Workflow and Documentation - Added support for PI Web API data ingestion, including real-time and historical data retrieval. - Implemented timestamp normalization to ensure consistency across records. - Updated README.md to reflect new features and detailed workflow execution flow. - Enhanced API class with comprehensive error handling and cleanup operations. - Improved CoreScouter workflow with early exit conditions and detailed processing steps. - Updated configuration parameters for better clarity and usability.
121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
from temporalio import workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from datetime import timedelta
|
|
from typing import Any
|
|
|
|
from sientia_do.temporal.policies import retry_policy
|
|
|
|
from scouter.activities.activities import Activities
|
|
|
|
|
|
@workflow.defn(name='scouter')
|
|
class Scouter:
|
|
"""
|
|
Main Scouter workflow that orchestrates data ingestion and processing.
|
|
|
|
This workflow serves as the entry point for data processing pipelines. It loads
|
|
data from MongoDB collections, manages data timestamps for incremental processing,
|
|
and delegates the actual data processing to the CoreScouter workflow.
|
|
|
|
The workflow implements a robust data ingestion pattern with:
|
|
- Incremental data loading based on last processed timestamp
|
|
- Automatic timestamp management for data continuity
|
|
- Error handling and retry policies
|
|
- Child workflow orchestration for data processing
|
|
"""
|
|
|
|
@workflow.run
|
|
async def run(self, input_data: dict[str, Any]) -> None:
|
|
"""
|
|
Execute the main Scouter workflow.
|
|
|
|
This method orchestrates the complete data ingestion process:
|
|
1. Retrieves the last processed timestamp from Redis
|
|
2. Loads new data from MongoDB since the last timestamp using collection name
|
|
format: `raw_{schedule_name}`
|
|
3. Updates the last processed timestamp with the most recent data point
|
|
4. Delegates data processing to the CoreScouter workflow
|
|
|
|
If no new data is found in MongoDB, the workflow exits early without updating
|
|
the timestamp or invoking the CoreScouter workflow.
|
|
|
|
Args:
|
|
input_data (dict[str, Any]): Configuration and parameters for the workflow execution.
|
|
Required fields:
|
|
- topic (str): The Kafka topic name for data source identification
|
|
- schedule_name (str): Unique identifier for the data collection schedule
|
|
- model_name (str): Name of the data model being processed
|
|
- model_id (str): Unique identifier for the data model
|
|
- trigger_laborious (bool): Flag to enable intensive data processing
|
|
- filters (dict[str, str]): Data quality filters configuration
|
|
- schema (str): Target database schema for data export
|
|
- table_name (str): Target table name for data export
|
|
- retention_time (int): Data retention period in Redis (seconds)
|
|
- model_tags (dict[str, Any]): Tag-specific configuration including:
|
|
- data_range: [min, max] values for data validation
|
|
- aggr_function: Aggregation method (avg, mdn, max, min, lts)
|
|
- frequency: Data collection frequency in milliseconds
|
|
- topics: List of Kafka topics for data routing
|
|
|
|
Returns:
|
|
None: This workflow doesn't return data, it orchestrates data processing
|
|
|
|
Raises:
|
|
WorkflowExecutionError: If workflow execution fails
|
|
ActivityExecutionError: If any activity fails after retry attempts
|
|
"""
|
|
|
|
input_data['workflow_name'] = 'scouter'
|
|
|
|
metadata = {
|
|
'metadata': {
|
|
'model_id': input_data['model_id'],
|
|
'model_name': input_data['model_name'],
|
|
'schedule_name': input_data['schedule_name'],
|
|
'workflow_name': input_data['workflow_name'],
|
|
}
|
|
}
|
|
|
|
last_data_timestamp = await workflow.execute_local_activity_method(
|
|
Activities.get_last_data_timestamp,
|
|
{
|
|
**metadata,
|
|
'workflow_name': input_data['workflow_name'],
|
|
'schedule_name': input_data['schedule_name'],
|
|
},
|
|
start_to_close_timeout=timedelta(seconds=60),
|
|
retry_policy=retry_policy,
|
|
)
|
|
|
|
data = await workflow.execute_local_activity_method(
|
|
Activities.load_latest_data,
|
|
{
|
|
**metadata,
|
|
'collection_name': f'raw_{input_data["schedule_name"]}',
|
|
'last_data_timestamp': last_data_timestamp,
|
|
},
|
|
start_to_close_timeout=timedelta(seconds=60),
|
|
retry_policy=retry_policy,
|
|
)
|
|
|
|
if not data:
|
|
return
|
|
|
|
await workflow.execute_activity_method(
|
|
Activities.put_last_data_timestamp,
|
|
{
|
|
**metadata,
|
|
'data': data,
|
|
'workflow_name': input_data['workflow_name'],
|
|
'schedule_name': input_data['schedule_name'],
|
|
},
|
|
start_to_close_timeout=timedelta(seconds=60),
|
|
retry_policy=retry_policy,
|
|
)
|
|
|
|
input_data['data'] = data
|
|
input_data['metadata'] = metadata
|
|
|
|
await workflow.execute_child_workflow('subworkflow.core_scouter', input_data)
|