# Processing unit ## Overview At its simplest form, a processing unit provides two key functions; a **stream** processing pipeline and a **metrics engine** that generate metrics based on a time schedule policy. This forms the **core of data handling** in Joule. The processing unit is composed of two main components: 1. [**Stream processors**](/joule/components/processors.md)\ A configurable series of processors that perform transformations, filters, and aggregations on event data. Pipelines are built from Joule’s **set of core processors** and allow users to construct sophisticated processing workflows. 2. [**Metrics engine**
](/joule/components/processors/enrichment/metrics.md)\ This engine calculates **complex metrics on a defined schedule** using SQL-based queries. Metrics are generated and updated at regular intervals, allowing for real-time insights into data. ### Example ```yaml stream: ... processing unit: pipeline: # Here comes the list of processors metrics engine: # Here comes the metrics declarations ``` ## Pipeline A pipeline is a sequence of processors that compute functions on events. Joule provides out-of-the-box a set of core processors to enable you to build useful use cases, learn more about the available processors and analytic tools.


Processors	Processors are the core of the Joule platform, each performing a specific task. These create use case when linked together		/pages/21H6be81tCxZDwGJDYxq
Analytics tools	Define math expressions or provide as a file using Joule supported languages and APIs		/pages/y5uZ4PFaYPXNzCR7iNXs

### Example This example defines a pipeline that filters, groups, and aggregates events: 1. **Filter**\ Filters events that match where `symbol != 'A'`. 2. **Tumbling time window**\ Groups events into 5-second windows. 3. **Aggregations**\ Calculates `MIN` and `MAX` values for `ask` and `bid` fields, grouped by `symbol`. 4. **Event emission**\ Outputs one event per `symbol` with aggregated values at the end of each window. ```yaml stream: ... pipeline: - filter: expression: "symbol != 'A'" - time window: emitting type: tumblingQuoteAnalytics aggregate functions: MIN: [ask, bid] MAX: [ask, bid] policy: type: tumblingTime window size: 5000 ``` ## Metrics engine The metrics engine provides the ability to compute complex metrics based upon SQL queries, the embedded SQL engine provides. {% hint style="info" %} This feature which is enabled by default {% endhint %} ### Example ```yaml stream: ... metrics engine: policy: timeUnit: MINUTES frequency: 1 startup delay: 2 foreach metric compute: metrics: - name: BidMovingAverage metric key: symbol table definition: standardQuoteAnalyticsStream.BidMovingAverage (symbol VARCHAR, avg_bid_min FLOAT, avg_bid_avg FLOAT,avg_bid_max FLOAT) query: SELECT symbol, AVG(bid_MAX) AS mt_avg_bid_min, AVG(bid_MAX) AS mt_avg_bid_max OVER ( PARTITION BY symbol ORDER BY eventTime ASC RANGE BETWEEN INTERVAL 1 SECONDS PRECEDING AND INTERVAL 1 SECONDS FOLLOWING) AS 'bid max Moving Average' FROM quotesStream.tumblingQuoteAnalytics_View ORDER BY 1; truncate on start: true compaction: frequency: 8 timeUnit: HOURS ``` ### policy The `policy` block defines the schedule for executing metrics calculations and managing data within the `metrics engine`. The following table describes each attribute.

Attribute	Description	Required	Default	Supported values
policy	Defines the scheduling policy for metric calculations, setting the timing and intervals for updates	true	N/A	N/A
timeUnit	Sets the time unit used for the `frequency` and `startup delay` intervals	true	MINUTES	SECONDS, MINUTES, HOURS
frequency	Specifies how often metric calculations are performed	false	1 Minute	Any positive integer in `timeUnit`
startup delay	Delay before the initial metric calculation	false	5 Minutes	Any positive integer in `timeUnit`

### Metrics computations The `foreach metric compute` syntax defines a metric table, computation, management and assigns it to a named metric family. The following table describes each attribute. | Attributes | Description | Required | Default | Supported values | | --------------------- | ------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------- | ------- | ----------------------- | | **name** | A unique identifier for the metric; also referred to as the metrics family name |

| N/A | N/A | | **metric key** | Generates optimised metric queries for user lookups and management functions |

| N/A | N/A | | **table definition** | Defines the SQL table for storing and accessing metrics, including the schema as part of the table name |

| N/A | N/A | | **query** | An ANSI SQL query executed periodically, with results inserted into the defined metric table |

| N/A | N/A | | **Management** | Default management processes on startup for efficient memory and housekeeping. Enabled hourly |

| N/A | N/A | | **truncate on start** | Truncates the table on startup if set to `true` |

| HOURS | SECONDS, MINUTES, HOURS | --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.fractalworks.io/joule/components/pipelines/processing-unit.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.