Stream Processing with SQL
Relevant source files
The following files were used as context for generating this wiki page:
The Stream Processor is a core engine in FluentDo Agent that allows for real-time data analysis and manipulation using SQL queries. It enables users to define processing tasks that can aggregate, filter, and transform records as they flow through the pipeline. This is particularly useful for scenarios like counting HTTP status codes, calculating average values, or selecting specific records based on complex conditions before they are sent to their final destination.
The engine is designed to work with different input and output plugins. It can be attached to any input plugin to process its data. The results of the SQL queries are then routed to one or more output plugins. The primary mechanism for using this feature is through a combination of the in_stream_processor input plugin and the processor_sql processor plugin.
Sources: source/src/stream_processor/README.md
Architecture
The Stream Processing subsystem consists of three main components: the Stream Processor Engine, the Stream Processor Input Plugin (in_stream_processor), and one or more Processor plugins (e.g., processor_sql).
- Stream Processor Engine (
flb_sp): The central component that manages stream processing tasks. It is responsible for creating streams, registering tasks (SQL queries), processing incoming data chunks, and routing the results. - Input Plugin (
in_stream_processor): A dedicated input plugin that acts as a bridge. It receives data from other input plugins (via routing) and feeds it into the Stream Processor engine. - Processor Plugin (
processor_sql): This plugin executes the actual SQL queries. It is instantiated for each task defined in the configuration and processes the records passed to it by the engine.
The diagram below illustrates the high-level architecture and data flow.
This diagram shows how data flows from standard input plugins, is routed to the in_stream_processor, processed by the Stream Processor Engine using the processor_sql plugin, and finally sent to output plugins.
Sources: source/src/stream_processor/README.md, source/src/stream_processor/flb_sp.c:1017-1038
Configuration
To enable Stream Processing, you need to configure [SERVICE], [INPUT], [STREAM_PROCESSOR], and [OUTPUT] sections in your configuration file.
1. Service Configuration
The streams.conf file must be loaded in the main [SERVICE] block.
[SERVICE]
Flush 1
Daemon Off
Log_Level info
streams_file streams.conf
Sources: source/src/stream_processor/README.md
2. Stream Processor Configuration (streams.conf)
This file defines the stream processing tasks. Each task is a [STREAM_PROCESSOR] section that specifies an input stream, a name, and an SQL query to execute.
A [STREAM_PROCESSOR] task section supports the following keys:
| Key | Description | Example |
|---|---|---|
Name | A unique name for the stream processing task. | my_query |
Exec | The SQL query to be executed on the incoming stream. | SELECT * FROM STREAM:my_stream WHERE value > 10; |
Tag | A tag to be applied to the records created by this task. | processed_data |
Window | Specifies the window type and size for aggregations (e.g., tumbling). | tumbling(5s) |
Sources: source/src/stream_processor/flb_sp.c:830-880, source/src/stream_processor/README.md
Example streams.conf:
[STREAM_PROCESSOR]
Name my_app_logs
Exec SELECT * FROM STREAM:my_stream WHERE app = 'fluent-bit';
[STREAM_PROCESSOR]
Name http_codes
Exec SELECT tag, code, count(*) FROM STREAM:my_stream GROUP BY tag, code;
Sources: source/src/stream_processor/README.md
3. Input and Routing
You need to configure an in_stream_processor input plugin and route data from other inputs to it.
[INPUT]
Name dummy
Tag my_stream
Dummy {"app": "fluent-bit", "value": 15}
# Route data from 'my_stream' tag to the stream processor
[INPUT]
Name stream_processor
Tag my_stream
stream my_stream
The stream_processor input plugin takes the stream property, which must match the stream name used in the FROM STREAM: clause of your SQL queries.
Sources: source/plugins/in_stream_processor/sp.c:183-205
Data Flow and Processing Logic
The processing of a data chunk follows a specific sequence of operations within the engine.
This sequence shows how a data chunk is ingested, routed to the stream processor, processed by an SQL task, and how the results are re-injected into the pipeline.
Sources: source/plugins/in_stream_processor/sp.c:100-120, source/src/stream_processor/flb_sp.c:1017-1038
Key Functions
| Function | Description | Source File |
|---|---|---|
flb_sp_create() | Initializes the stream processor context. | source/src/stream_processor/flb_sp.c:980 |
flb_sp_do() | Main entry point to process an incoming data chunk. It iterates through tasks and executes them. | source/src/stream_processor/flb_sp.c:1017 |
flb_sp_task_create() | Creates a new stream processing task, linking it to a processor plugin like processor_sql. | source/src/stream_processor/flb_sp.c:489 |
cb_sql_init() | Initialization callback for the processor_sql plugin. It parses the Query configuration key. | source/plugins/processor_sql/sql.c:303 |
cb_sql_process() | Processing callback for processor_sql. It executes the SQL query against the input records. | source/plugins/processor_sql/sql.c:196 |
sp_in_callback() | The callback for in_stream_processor that receives data and passes it to flb_sp_do(). | source/plugins/in_stream_processor/sp.c:100 |
Core Data Structures
The stream processing engine relies on several key data structures to manage its state and operations.
Sources: source/include/fluent-bit/flb_sp.h
This diagram shows the relationship between the main stream processor context (flb_sp), its tasks (flb_sp_task), and the streams it monitors. Each task has a context, such as flb_processor_sql_ctx for SQL processing.
struct flb_sp
The main context for the Stream Processor engine. It holds lists of all configured tasks and streams.
Sources: source/include/fluent-bit/flb_sp.h:129-141
struct flb_sp_task
Represents a single processing task, typically defined by a [STREAM_PROCESSOR] section. It contains the task's name, the SQL query (exec), and an instance of the processor plugin that will execute the query.
Sources: source/include/fluent-bit/flb_sp.h:108-121
struct flb_processor_sql_ctx
The context for an sql processor instance. It holds the parsed SQL command (flb_sp_cmd) which is generated from the Exec query string.
Sources: source/plugins/processor_sql/sql.c:56-61
struct flb_sp_cmd
Represents a parsed SQL command. It contains information about the SELECT columns, FROM stream, WHERE conditions, and GROUP BY clauses. This structure is the output of the SQL parser and is used by the processor_sql plugin to evaluate records.
Sources: source/include/fluent-bit/flb_sp.h:70-90
SQL Query Execution
The processor_sql plugin is responsible for parsing and executing SQL queries.
Initialization (cb_sql_init)
When the plugin is initialized, it performs the following steps:
- Retrieves the
Queryproperty from the configuration. - Calls
flb_sp_parser()to parse the SQL query string into anflb_sp_cmdstruct. - Stores the parsed command in the plugin's context (
flb_processor_sql_ctx).
If the query is invalid, initialization fails.
Sources: source/plugins/processor_sql/sql.c:303-333
Processing (cb_sql_process)
For each data chunk, the cb_sql_process function is invoked. It iterates through each record in the chunk and applies the logic from the parsed flb_sp_cmd:
- WHERE clause: If a
WHEREcondition exists, it is evaluated for the current record. If the condition is false, the record is skipped. - Aggregation: If the query is an aggregation (contains
GROUP BYor aggregation functions likeCOUNT(*)), the record is processed by the aggregation engine. - Record Projection: If the query is a simple
SELECT, a new record is created based on the selected fields. - Result Set: The function returns a new set of records representing the query's results.
Sources: source/plugins/processor_sql/sql.c:196-291
Summary
The Stream Processing with SQL feature provides a powerful, integrated way to perform real-time data analysis within the FluentDo Agent pipeline. By defining simple SQL queries, users can filter, aggregate, and transform data streams without needing external processing systems. The architecture, based on a central engine and specialized plugins, offers a flexible and efficient mechanism for handling complex data manipulation tasks directly on the agent.