Anomaly Detection with Confluent Manager for Apache Flink

Anomaly detection is an important use-case for Apache Flink®. You can deploy anomaly detection jobs using Confluent Manager for Apache Flink (CMF), a central management component of Confluent Platform for Apache Flink that enables users to securely manage a fleet of Flink jobs across environments.

How does anomaly detection work?

Anomaly detection uses statistical and machine learning techniques to identify data points, events, or observations that deviate significantly from the expected pattern within a stream. These deviations, or anomalies, can indicate critical incidents, like system failures, fraudulent activities, or unusual behavioral patterns that require further investigation.

Built on SQL Forecast, anomaly detection identifies unexpected deviations in real-time, helping improve data quality, detect operational issues instantly, and enable faster decision-making.

Flink integrates anomaly detection capabilities on your streams by using the ML_DETECT_ANOMALIES function. This function leverages a popular machine learning algorithm, named ARIMA (Autoregressive Integrated Moving Average), that is optimized for real-time performance, enabling accurate and reliable anomaly detection.

Confluent Platform for Apache Flink also provides AI_DETECT_ANOMALIES, which detects anomalies with pretrained time-series foundation models that run in model servers in your Kubernetes cluster. It needs no model parameters and scores values from the 21st value of a series onward by default. For details, see Detect anomalies with AI_DETECT_ANOMALIES.

Your data must include:

• A timestamp column.

• A target column representing some quantity of interest at each timestamp.

Understand the ARIMA model

ARIMA, which stands for Autoregressive Integrated Moving Average, is a powerful statistical technique used for time series analysis and forecasting. It combines three key components to model and predict future values based on past data:

• Autoregressive (AR): This component uses past values of the time series to predict future values. It assumes that current observations are influenced by previous observations.

• Integrated (I): This refers to the differencing of raw observations to make the time series stationary. Stationary data has consistent statistical properties over time, which is crucial for accurate forecasting.

• Moving Average (MA): This component incorporates the relationship between an observation and a residual error from a moving average model applied to previous observations.

ARIMA models are used widely in various fields, including finance, economics, and weather forecasting, to analyze time-dependent data and make predictions. They are particularly effective when dealing with non-seasonal time series data that exhibits trends or patterns over time.

The model is typically represented as ARIMA(p,d,q), where:

• p: The number of autoregressive terms

• d: The degree of differencing

• q: The order of the moving average

By adjusting these parameters, analysts can fine-tune the model to best fit their specific time series data and improve forecasting accuracy.

Be aware that forecasting accuracy can vary greatly, based on many parameters, and there is no guarantee of correctness in predictions made by using ARIMA and machine learning.

Anomaly detection parameters

The following parameters are supported for anomaly detection.

Model configuration

Use the following parameters to configure the ML_DETECT_ANOMALIES function. Pass them as a JSON string in the optional third argument, ML_DETECT_ANOMALIES(value, timestamp[, config]), for example with JSON_OBJECT. If you omit the configuration argument, the function uses the defaults below, except that enableStl is TRUE.

enableStl

• Default: FALSE, or TRUE if you omit the configuration argument

Enable STL (Seasonal-Trend decomposition using Loess smoothing) decomposition. If set to TRUE, the time series is decomposed into trend, seasonal, and remainder components.

STL is a statistical method that separates a time series into three additive components: trend, seasonal, and remainder.

• The trend component captures the long-term direction or movement in the data.

• The seasonal component captures regular, predictable patterns that repeat over fixed periods.

• The remainder component captures random fluctuations that aren’t explained by trend or seasonality.

This decomposition helps improve forecasting accuracy by enabling the ARIMA model to better analyze and predict the underlying patterns in your data.

evalWindowSize

• Default: 5

• Validation: Must be >= 1 and <= 10000

Keeps the last evalWindowSize squared errors for both RMSE and AIC calculations. RMSE uses the rolling sum of these errors, while AIC uses the count (n) and sum of those same errors. A smaller window makes both metrics more responsive to recent errors; a larger window smooths out short-term noise.

An error is recorded only when a row arrives at the timestamp the model forecast, which is the previous row’s timestamp plus the most common interval between rows, and a row flagged as an anomaly counts as zero error. If your timestamps are unevenly spaced, rmse and aic can stay NULL.

m

• Default: If enableStl is TRUE, seasonality is estimated using ACF.

• Validation: m >= 0 and m <= minTrainingSize / 2. Setting m requires enableStl to be TRUE.

The seasonal period length. If enableStl is TRUE, the model uses an ACF (Autocorrelation Function) to estimate the seasonal period length. The ACF measures the correlation between a time series and its lagged versions at different time intervals. This helps to identify repeating patterns and seasonal cycles in the data by detecting how strongly current values relate to past values at various lags.

maxTrainingSize

• Default: 512

• Validation: minTrainingSize <= maxTrainingSize <= 10000

Maximum number of historical data points used for training. This value limits the training window to recent data.

minTrainingSize

• Default: 128

• Validation: 1 <= minTrainingSize <= maxTrainingSize

Minimum number of data points required for training. The model won’t train until the quantity of available data is greater than this value.

p

• Default: NULL

• Validation: If provided, must be >= 0

Order of the AR (Autoregressive) term in the ARIMA model, which defines the number of lagged values used in the model. This value represents the number of past values that directly influence the current prediction.

If p is not provided, the model uses auto-ARIMA to determine the best order.

q

• Default: NULL

• Validation: If provided, must be >= 0

Order of the MA (Moving Average) term in the ARIMA model, which defines the number of lagged error terms in the model. This value represents the number of past prediction errors that influence the current prediction.

If q is not provided, the model uses auto-ARIMA to determine the best order.

d

• Default: NULL

• Validation: If provided, must be >= 0

Order of differencing in the ARIMA model, which defines the number of times the data is differenced to achieve stationarity. A value of 0 means no differencing, 1 means first difference, and so on.

If d is not provided, the model uses auto-ARIMA to determine the best order.

updateInterval

• Default: 1

• Validation: Must be > 0

Frequency of model updates. This value defines how often the model retrains with new data.

confidencePercentage

• Default: 99.0

• Validation: 1 <= value <= 100. Pass a decimal value, such as 95.0; an integer value such as 95 fails. A value of 100 is treated as 99.9999.

The desired confidence level for computing anomaly bounds.

Anomaly detection results

The ML_DETECT_ANOMALIES function returns a row with the following fields, in the order timestamp, actual_value, forecast_value, lower_bound, upper_bound, is_anomaly, rmse, and aic. Until the model has trained and made its first forecast, the function sets only timestamp and actual_value, and the other fields are NULL.

actual_value

• Type: DOUBLE

The input value at the timestamp.

aic

• Type: DOUBLE

The AIC (Akaike Information Criterion) at the time the forecast was generated, calculated over the last evalWindowSize errors.

AIC is a statistical measure used to evaluate and compare the quality of different models. It balances model accuracy against complexity by penalizing models with more parameters. Lower AIC values indicate better model performance, which helps to identify the optimal model configuration that provides the best fit to the data without overfitting.

forecast_value

• Type: DOUBLE

The forecast value at the timestamp.

lower_bound

• Type: DOUBLE

The lower confidence interval for the forecasted value.

rmse

• Type: DOUBLE

The RMSE (Root Mean Square Error) at the time the forecast was generated, calculated over the last evalWindowSize errors.

RMSE measures the difference between the predicted and actual values in a time series by calculating the square root of the average of the squared differences between the predicted and actual values. A lower RMSE indicates better model performance, because it means the model’s predictions are closer to the actual values.

timestamp

• Type: TIMESTAMP

The LocalDateTime for this forecast.

upper_bound

• Type: DOUBLE

The upper confidence interval for the forecasted value.

is_anomaly

• Type: BOOLEAN

TRUE if actual_value falls below lower_bound or above upper_bound; otherwise FALSE. NULL until the model has trained.

Example

1. Create a test table that has a timestamp field for your time series data.

   CREATE TABLE orders (
       total_orderunits double,
       summed_ts TIMESTAMP(3) NOT NULL,
       WATERMARK FOR summed_ts AS summed_ts
   );
   

2. Insert mock time series data into the table.

   INSERT INTO orders (total_orderunits, summed_ts) VALUES
   (102.12, TIMESTAMP '2024-11-19 10:00:00.000'),
   (103.45, TIMESTAMP '2024-11-19 10:01:00.000'),
   (101.89, TIMESTAMP '2024-11-19 10:02:00.000'),
   (104.23, TIMESTAMP '2024-11-19 10:03:00.000'),
   (102.78, TIMESTAMP '2024-11-19 10:04:00.000'),
   (102.12, TIMESTAMP '2024-11-19 10:05:00.000'),
   (103.45, TIMESTAMP '2024-11-19 10:06:00.000'),
   (101.89, TIMESTAMP '2024-11-19 10:07:00.000'),
   (104.23, TIMESTAMP '2024-11-19 10:08:00.000'),
   (102.78, TIMESTAMP '2024-11-19 10:09:00.000'),
   (102.12, TIMESTAMP '2024-11-19 10:10:00.000'),
   (103.45, TIMESTAMP '2024-11-19 10:11:00.000'),
   (101.89, TIMESTAMP '2024-11-19 10:12:00.000'),
   (104.23, TIMESTAMP '2024-11-19 10:13:00.000'),
   (102.78, TIMESTAMP '2024-11-19 10:14:00.000'),
   (102.12, TIMESTAMP '2024-11-19 10:15:00.000'),
   (103.45, TIMESTAMP '2024-11-19 10:16:00.000'),
   (101.89, TIMESTAMP '2024-11-19 10:17:00.000'),
   (104.23, TIMESTAMP '2024-11-19 10:18:00.000'),
   (102.78, TIMESTAMP '2024-11-19 10:19:00.000'),
   (102.12, TIMESTAMP '2024-11-19 10:20:00.000'),
   (103.45, TIMESTAMP '2024-11-19 10:21:00.000'),
   (101.89, TIMESTAMP '2024-11-19 10:22:00.000'),
   (104.23, TIMESTAMP '2024-11-19 10:23:00.000'),
   (102.78, TIMESTAMP '2024-11-19 10:24:00.000'),
   (102.12, TIMESTAMP '2024-11-19 10:25:00.000'),
   (103.45, TIMESTAMP '2024-11-19 10:26:00.000'),
   (101.89, TIMESTAMP '2024-11-19 10:27:00.000'),
   (104.23, TIMESTAMP '2024-11-19 10:28:00.000'),
   (102.78, TIMESTAMP '2024-11-19 10:29:00.000');
   

1. Run the following statement to perform anomaly detection on the mock data in the orders table.

   SELECT
       ML_DETECT_ANOMALIES(total_orderunits, summed_ts, JSON_OBJECT('p' VALUE 1, 'q' VALUE 1, 'd' VALUE 1, 'minTrainingSize' VALUE 10, 'enableStl' VALUE false))
       OVER (
           ORDER BY summed_ts
           RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS anomaly
   FROM orders;
   

2. Run the following statement to perform anomaly detection with AutoArima on the mock data in the orders table.

   SELECT
       ML_DETECT_ANOMALIES(total_orderunits, summed_ts, JSON_OBJECT('minTrainingSize' VALUE 10, 'enableStl' VALUE false))
       OVER (
           ORDER BY summed_ts
           RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS anomaly
   FROM orders;
   

3. If there are multiple values for the same timestamps, or the timestamps are not uniform, use tumble windows with ML_DETECT_ANOMALIES.

   CREATE TABLE orderunits_result_anomaly DISTRIBUTED INTO 1 BUCKETS AS
   WITH windowed_avg AS (
     SELECT
       window_start,
       window_end,
       window_time,
       AVG(total_orderunits) AS avg_total_orderunits
     FROM TABLE(
       TUMBLE(TABLE orders, DESCRIPTOR(summed_ts), INTERVAL '1' MINUTE)
     )
     GROUP BY window_start, window_end, window_time
   )
   SELECT
     window_start,
     avg_total_orderunits,
     ML_DETECT_ANOMALIES(avg_total_orderunits, window_time, JSON_OBJECT('minTrainingSize' VALUE 10, 'enableStl' VALUE false))
       OVER (
         ORDER BY window_time
         RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
       ) AS anomaly
   FROM windowed_avg;
   

Detect anomalies with AI_DETECT_ANOMALIES

Important

AI_DETECT_ANOMALIES is a preview feature. Interfaces might change in future releases.

AI_DETECT_ANOMALIES flags the values of a series that fall outside the range a time-series foundation model predicts for them. After each value, the function asks the model to forecast the next value with a range. When the next value arrives, the function compares it with that range. The models are pretrained, so the function needs no parameters and no training data. With the default settings, it scores values from the 21st value of a series onward.

The models run in model servers in your Kubernetes cluster. Before you can call the function, an administrator must deploy the model servers and register the function in your environment, as described in Set Up AI Functions in Confluent Manager for Apache Flink.

Syntax

AI_DETECT_ANOMALIES(value, timestamp[, config])
  OVER (ORDER BY timestamp RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

value

The value of the series, as a DOUBLE. The function leaves rows whose value is NULL, NaN, or infinite out of the series.

timestamp

The time of the value, as a TIMESTAMP(3). The function takes the series’ step from the median interval between timestamps, and forecasts the value at the timestamp of the last value plus one step. It scores only a row that arrives at that timestamp. For any other row, it returns the result of the last row it scored. If the timestamps of a series are not evenly spaced, aggregate the series into fixed windows first, for example with TUMBLE, as in the last step of the ML_DETECT_ANOMALIES example.

config

Optional. A JSON object, as a string or from JSON_OBJECT, with the settings in the following table. The configuration must be the same for every row of a series.

Call the function from a statement whose current catalog is the environment catalog, _env_<environment>, and whose current database is default. To score several series in one statement, add PARTITION BY to the OVER clause.

Configuration

Key	Default	Description
model	ttm	The model to forecast with: ttm, flowstate, or patchtst. The environment must have a model server for it.
confidencePercentage	99.0	The confidence level of the range, greater than 0 and less than 100. For how the model limits the range, see Ranges and models.
lowerBoundConfidencePercentage	None	The confidence that a value is above the lower bound, greater than 0 and less than 100. Overrides confidencePercentage for the lower bound.
upperBoundConfidencePercentage	None	The confidence that a value is below the upper bound, greater than 0 and less than 100. Overrides confidencePercentage for the upper bound.
minContextSize	20	The number of values the function waits for before it forecasts. At least 2.
maxContextSize	512	The largest number of recent values that the function sends to the model. At least minContextSize.
rmseWindowSize	5	The number of recent forecasts that rmse is computed over. At least 1.

Keys are case-insensitive. If you set only one of the two bound keys, also set confidencePercentage, which then sets the other bound. Otherwise the statement fails.

Ranges and models

The function turns the confidence level into two percentiles, for example the 0.5th and the 99.5th percentile for 99.0. It then takes the lower and upper bound from the percentiles of the model’s forecast that are closest to them. The models forecast different percentiles:

• TTM-R3 and FlowState forecast the 10th to the 90th percentile, in steps of 10. Their range is at most the 80% range, so any confidencePercentage of 80 or more gives the same range.

• PatchTST-FM forecasts the 1st to the 99th percentile, so its range is at most the 98% range.

A narrower range flags more values. With the series in the following example, FlowState forecasts narrower ranges than the other two models and flags many more values.

Results

AI_DETECT_ANOMALIES returns a row with the following fields:

Field	Type	Description
timestamp	TIMESTAMP(3)	The timestamp of the scored value.
actual_value	DOUBLE	The scored value.
forecast_value	DOUBLE	The value that the model forecast for this timestamp.
lower_bound	DOUBLE	The lower bound of the range.
upper_bound	DOUBLE	The upper bound of the range.
is_anomaly	BOOLEAN	TRUE if actual_value is below lower_bound or above upper_bound, otherwise FALSE.
rmse	DOUBLE	The root mean square error of the last rmseWindowSize forecasts. A value flagged as an anomaly counts as an error of zero, so that anomalies do not raise rmse.
aic	DOUBLE	Always NULL. The result has the same fields as ML_DETECT_ANOMALIES.

Until the function has scored a value, only timestamp and actual_value are set, and the other fields are NULL.

Example

1. Create a table that generates a series with one value per second, in the environment catalog:

   CREATE TABLE sensor_readings (
     id BIGINT,
     ts AS CAST(TIMESTAMPADD(SECOND, CAST(id AS INT), TIMESTAMP '2026-01-01 00:00:00') AS TIMESTAMP(3)),
     WATERMARK FOR ts AS ts
   ) WITH (
     'connector' = 'datagen',
     'fields.id.kind' = 'sequence',
     'fields.id.start' = '1',
     'fields.id.end' = '10000',
     'rows-per-second' = '20'
   );
   

2. Score a wave with a period of 24 seconds that jumps to 30 every 50 seconds:

   SELECT
     id,
     v,
     a.forecast_value,
     a.lower_bound,
     a.upper_bound,
     a.is_anomaly
   FROM (
     SELECT
       id,
       v,
       AI_DETECT_ANOMALIES(v, ts, JSON_OBJECT('model' VALUE 'ttm'))
         OVER (ORDER BY ts RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS a
     FROM (
       SELECT
         id,
         ts,
         CAST(CASE WHEN MOD(id, 50) = 0 THEN 30.0
              ELSE 10 + 5 * SIN(2 * PI() * id / 24) + 0.3 * SIN(id * 1.7) END AS DOUBLE) AS v
       FROM sensor_readings
     )
   );
   

   The first 20 rows have no forecast. The rows where the series jumps, with id 50, 100, 150, and so on, have is_anomaly set to TRUE. A few other rows can be flagged as well, for example the row after a jump.

Limitations

• Each row sends one request to the model server and waits for the answer, so a statement processes rows only as fast as its model server answers.

• timesfm-2.5 is not available, because no model server for it is provided.

For the errors that the function reports, see the troubleshooting section of Set Up AI Functions in Confluent Manager for Apache Flink.

Related content

• Set Up AI Functions in Confluent Manager for Apache Flink

• Forecast Data Trends with Confluent Platform for Apache Flink

• Submit Flink SQL Statement with Confluent Manager for Apache Flink

• Deploy and Manage Statements in Confluent Manager for Apache Flink

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.