The sample file contains JSON records. Run the following command to append the parsers.conf file and instruct the Tail input plugin to parse content as JSON:
This command prints the parsed content to the standard output interface. The parsed content includes a tag associated with each record and an array with two fields: a timestamp and a record map:
3. Select specific record keys
Run the following command to create a stream processor query using the -T flag and change the output to the Null plugin. This obtains the stream processing results in the standard output interface and avoids confusion in the terminal.
The previous query aims to retrieve all records for which the country key contains the value Chile. For each match, it composes and outputs a record that only contains the keys word and num:
4. Calculate average value
Run the following command to use the AVG aggregation function to get the average value of ingested records:
The previous query yields the following output:
The resulting output contains multiple records because Fluent Bit processes data in chunks, and the stream processor processes each chunk independently. To process multiple chunks at the same time, you can group results using time windows.
5. Group results and windows
Grouping results within a time window simplifies data processing. Run the following command to group results by country and calculate the average of num with a one-second processing window:
The previous query yields the following output:
6. Ingest stream processor results as a new stream of data
Next, instruct the stream processor to ingest results as part of the Fluent Bit data pipeline and assign a tag to each record.
Run the following command, which uses a CREATE STREAM statement to tag results with the sp-results tag, then outputs records with that tag to standard output:
