Process IIoT Sensor Streams at the Edge with Bytewax and Polars

                      
                      
import os
from bytewax import Dataflow, run
from bytewax.dataflow import map, filter
import polars as pl

# Configuration
SENSOR_API_URL = os.getenv('SENSOR_API_URL')  # URL for IIoT sensor data

# Function to fetch sensor data
async def fetch_sensor_data() -> pl.DataFrame:
    try:
        # Simulated API call to fetch data
        # Replace with actual API call logic
        response = await fetch(SENSOR_API_URL)
        return pl.from_dict(response.json())  # Convert response to DataFrame
    except Exception as e:
        print(f"Error fetching sensor data: {e}")
        return pl.DataFrame()  # Return empty DataFrame on error

# Define the data processing flow
def process_data(data: pl.DataFrame) -> pl.DataFrame:
    # Perform data processing, e.g., filtering outliers
    return data.filter(pl.col("value") < 100)  # Example condition

# Main dataflow definition
def create_dataflow() -> Dataflow:
    df = Dataflow()  
    df.map(fetch_sensor_data)
    df.map(process_data)
    return df

# Execute the dataflow
if __name__ == '__main__':
    run(create_dataflow(), worker_count=4)  # Adjust the worker count for parallel processing