1. Project Overview & Quickstart (fluvio-community/fluvio)

File: README.md

Fluvio is a lean and mean distributed data streaming engine written in Rust. Combined with Stateful DataFlow distributed stream processing framework, Fluvio provides a unified composable distributed streaming and stream processing paradigm for developers.

Quick Start - Get started with Fluvio in 2 minutes or less!

Step 1. Download Fluvio Version Manager:

Fluvio is installed via the Fluvio Version Manager, shortened to fvm.

To install fvm, run the following command:

TEMPORARY NOTE: While we are transitioning to a new fluvio-community hosted
fluvio build and release, you should install the dev version of FVM:

bash

curl -fsS https://raw.githubusercontent.com/fluvio-community/fluvio/master/install.sh | FVM_VERSION=dev bash

After we get the fluvio-community based released sorted again, it will return to:

bash

curl -fsS https://raw.githubusercontent.com/fluvio-community/fluvio/master/install.sh | bash

As part of the initial setup, fvm will also install the Fluvio CLI available in the stable channel as of the moment of installation.

Fluvio is stored in $HOME/.fluvio, with the executable binaries stored in $HOME/.fluvio/bin.

For the best compatibliity on Windows, InfinyOn recommends WSL2

Step 2. Start a cluster:

Start cluster on you local machine with the following command:

bash

fluvio cluster start

Step 3. Create Topic:

The following command will create a topic called hello-fluvio:

bash

fluvio topic create hello-fluvio

Step 4. Produce to Topic, Consume From Topic:

Produce data to your topic. Run the command first and then type some messages:

bash

fluvio produce hello-fluvio
> hello fluvio
Ok!
> test message
Ok!

Consume data from the topic, Run the following command in a different terminal:

bash

fluvio consume hello-fluvio -B -d

Just like that! You have a local cluster running.

Using Pre-Built Fluvio Versions

You may want to use other Fluvio versions than the latest stable release. You can do so by specifying the version using environment variables.

Install Latest Dev Version

To install the latest development build of fvm and fluvio:

bash

curl -fsS https://raw.githubusercontent.com/fluvio-community/fluvio/master/install.sh | FVM_VERSION=dev bash

Install Specific Fluvio Version

bash

curl -fsS https://raw.githubusercontent.com/fluvio-community/fluvio/master/install.sh | FLUVIO_VERSION=x.y.z bash

Install Specific FVM Version

bash

curl -fsS https://raw.githubusercontent.com/fluvio-community/fluvio/master/install.sh | FVM_VERSION=v0.18.1 bash

Check Fluvio Core Documentation

Fluvio documentation will provide additional context on how to use the Fluvio clusters, CLI, clients, a development kits.

Check Stateful DataFlow Documentation

Stateful DataFlow designed to handle complex data processing workflows, allowing for customization and scalability through various programming languages and system primitives.

Learn how to build custom connectors

Fluvio can connect to practically any system that you can think of.

For first party systems, fluvio clients can integrate with the edge system or application to source data.
For third party systems fluvio connectors connect at the protocol level and collects data into fluvio topics.

Out of the box Fluvio has native http, webhook, mqtt, kafka inbound connectors. In terms of outbound connectors out of the box Fluvio supports http, SQL, kafka, and experimental builds of DuckDB, Redis, S3, Graphite etc.

Using Connector Development Kit, its intuitive to build connectors to any system fast.

Check out the docs and let us know if you need help building any connector.

Learn how to build custom smart modules

Fluvio applies wasm based stream processing and data transformations. We call these reusable transformation functions smart modules. Reusable Smart modules are built using Smart Module Development Kit and can be distributed using InfinyOn Cloud hub.

Clients

Fluvio Client API docs home

Language Specifc API docs:

Community Maintained:

Contributing

If you'd like to contribute to the project, please read our
Contributing guide.

Community

Many fluvio users and developers have made projects to share with the community.
Here a a few listed below:

Projects Using Fluvio

Swiftide Project: a Rust native library for building LLM applications
Real Time Stock Charts: See how Fluvio is used to update real time stock charts

Community Connectors

Qdrant Connector
Google Sheets Connector: Send data from Fluvio to Google Sheets
Elastic Connector: Send data from Fluvio to Elastic Search

Community Development Resources

More projects and utilities are available in the Fluvio Community Github Org

Hello World Fluvio Connector: Sample Fluvio connector template to build your own connector
Gurubase: Third-party AI/LLM Docs query

Contributors are awesome

Made with contrib.rocks.

Community

Many fluvio users and developers have made projects to share with the community.
Here a a few listed below:

Projects Using Fluvio

Real Time Stock Charts: See how Fluvio is used to update real time stock charts
Google Sheets Connector: Send data from Fluvio to Google Sheets
Elastic Connector: Send data from Fluvio to Elastic Search
Hello World Fluvio Connector: Sample Fluvio connector template to build your own connector

License

This project is licensed under the Apache license.

File: connector/docker/README.md

Introduction

This is sample project to demonstrate how to use the docker to run the connector.
This examples is for http connector but can be used for any connector.

To build image

To build docker image run:

text

make build

This will copy following files from your local file system:

your current fluvio profile
connector.yaml

The connector.yaml is connector config for http source

To run image

To run image, run:

text

make run

This setup will run host network mode. This means that connector will be able to access the local network. If you want to run in different network mode, you can modify the Makefile to change the network mode.

File: crates/fluvio-storage/README.md

Building storage cli

From root of the project

text

make build-storage-cli

and set alias for storage-cli

text

alias scli=$(PWD)/target/release/storage-cli

Running storage cli

text

scli log /tmp/fluvio/spu-logs-5001/t1-0/00000000000000000000.log

File: crates/fluvio-spu-schema/README.md

flv-api-spu

Fluvio API for SPU

License

This project is licensed under the Apache license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Fluvio by you, shall be licensed as Apache, without any additional
terms or conditions.

File: crates/fluvio-spu/README.md

Running SPU in development mode

To run SPU with default parameters.

text

./target/debug/fluvio-spu

Run server with debug flag

text

RUST_LOG=debug ./target/debug/fluvio-spu

File: crates/fluvio-socket/README.md

fluvio-protocol

Provide TCP socket wrapper for Fluvio protocol

build fluvio send utility

build fluvio-send utility

cargo build --all-features

invoking fluvio-send utility to send api request sample data

../target/debug/fluvio-send data/apirequest.bin

License

This project is licensed under the Apache license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Fluvio by you, shall be licensed as Apache, without any additional
terms or conditions.

File: crates/fluvio-smartmodule/README.md

Fluvio SmartModules

This crate provides types and macros for creating custom SmartModules,
which are WebAssembly modules that may be used to modify the behavior
of Fluvio streams to consumers. The currently supported SmartModule
types are filters, which may describe records to keep in or discard
from a stream.

Writing SmartModules

See the examples directory for full examples.

All SmartModules require adding crate-type = ['cdylib'] to the Cargo.toml.
For a quick setup using cargo-generate, see the SmartModule template.

toml

[package]
name = "fluvio-wasm-filter"
version = "0.1.0"
authors = ["Fluvio Contributors <[email protected]>"]
edition = "2024"

[lib]
crate-type = ['cdylib']

[dependencies]
fluvio-smartmodule = "0.8.0"

Filtering

For filtering, write your smartmodule using #[smartmodule(filter)] on your
top-level function. Consider this the "main" function of your SmartModule.

rust

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, Result};

#[smartmodule(filter)]
pub fn filter(record: &SmartModuleRecord) -> Result {
    let string = std::str::from_utf8(record.value.as_ref())?;
    Ok(string.contains('a'))
}

This filter will keep only records whose contents contain the letter a.

Mapping

Mapping functions use #[smartmodule(map)], and are also a top-level entrypoint.

rust

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, RecordData, Result};

#[smartmodule(map)]
pub fn map(record: &SmartModuleRecord) -> Result<(Option<RecordData>, RecordData)> {
    let key = record.key.clone();

    let string = std::str::from_utf8(record.value.as_ref())?;
    let int = string.parse::()?;
    let value = (int * 2).to_string();

    Ok((key, value.into()))
}

This SmartModule will read each input Record as an integer (i32), then multiply it by 2.

Aggregate

Aggregate functions are a way to combine the data from many input records.
Each time the aggregate function is called, it receives an "accumulated" value
as well as the value of the current record in the stream, and is expected to
combine the accumulator with the value to produce a new accumulator. This new
accumulator value will be passed to the next invocation of aggregate with
the next record value. The resulting stream of values is the output accumulator
from each step.

rust

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord, RecordData};

#[smartmodule(aggregate)]
pub fn aggregate(accumulator: RecordData, current: &SmartModuleRecord) -> Result<RecordData> {
    let mut acc = String::from_utf8(accumulator.as_ref().to_vec())?;
    let next = std::str::from_utf8(current.value.as_ref())?;
    acc.push_str(next);
    Ok(acc.into())
}

This SmartModule reads each record as a string and appends it to the accumulator string.

ArrayMap

ArrayMap functions are used to take one input record and create zero to many output records.
This can be used to chop up input records that logically represent more than one data point
and turn them into independent records. Below is an example where we take JSON arrays and
convert them into a stream of the inner JSON objects.

ignore

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, RecordData, Result};

#[smartmodule(array_map)]
pub fn array_map(record: &SmartModuleRecord) -> Result<Vec<(Option<RecordData>, RecordData)>> {
    // Deserialize a JSON array with any kind of values inside
    let array: Vec<serde_json::Value> = serde_json::from_slice(record.value.as_ref())?;

    // Convert each JSON value from the array back into a JSON string
    let strings: Vec<String> = array
        .into_iter()
        .map(|value| serde_json::to_string(&value))
        .collect::<core::result::Result<_, _>>()?;

    // Create one record from each JSON string to send
    let records: Vec<(Option<RecordData>, RecordData)> = strings
        .into_iter()
        .map(|s| (None, RecordData::from(s)))
        .collect();
    Ok(records)
}

Init

Init functions are optional but serve to configure any state the SmartModule requires at the beginning of its execution. Could be helpful for preparing the SmartModule's operational context. The example below demonstrates an init function that sets a key for the SmartModule to use:

ignore

use std::sync::OnceLock;

use fluvio_smartmodule::{
    smartmodule, Result, eyre,
    dataplane::smartmodule::{SmartModuleExtraParams, SmartModuleInitError},
};

static CRITERIA: OnceLock<String> = OnceLock::new();

#[smartmodule(init)]
fn init(params: SmartModuleExtraParams) -> Result<()> {
    if let Some(key) = params.get("key") {
        CRITERIA
            .set(key.clone())
            .map_err(|err| eyre!("failed setting key: {:#?}", err))
    } else {
        Err(SmartModuleInitError::MissingParam("key".to_string()).into())
    }
}

License

This project is licensed under the Apache license.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted
for inclusion in Fluvio by you, shall be licensed as Apache, without any additional
terms or conditions.

2. Official Technical Reference & Guides (fluvio-community/fluvio-docs)

File: README.md

Contributing

Prerequisites

NodeJS (v20.11.1 or higher)

Installation

bash

$ npm install

Development

Use npm start command to run the development server.

bash

$ npm start

[!NOTE]
The Search support is not available when running for development, if you want
to test the search functionality, you need to run the build and serve the
build directory.

Build Test

Build the solution for publishing and then serve bundle files using a static server.

bash

$ npm run build && npm run serve

The build command generates static content into the build directory and
can be served using any static contents hosting service.

Create Versions

Theres 2 sets of docs available in this repo that are versioned:

fluvio: The main documentation for Fluvio, which lives on docs dir
sdf: The documentation for SDF, which lives on sdf dir

Doc	Version Command
`fluvio`	`npm run docusaurus docs:version <VERSION>`
`sdf`	`npm run docusaurus docs:version:sdf <VERSION>`

[!WARNING]
Versions specified in the command should not be prefixed with v.

Release and Publish

In order to release a version and publish it you need to run:

bash

./bin/release

This will tag current version, push the tag to the repository, which triggers a
Release process and a Publish process.

Environments

Theres two environments for the documentation:

Production: The production environment is the default environment, it
contains the latest stable version of the documentation. This environment is
released when tagging a version.
Nightly: The nightly environment is the development environment, it
contains the latest changes merged into the main branch.

Dev tooling

If you are analyzing broken links, you can optionally use some external
link analyzers. The staps are tasks in the Justfile. This uses
the just task runner.

On macos: brew install just, on debian/ubuntu based linux: apt install just
Other environments https://github.com/casey/just#packages

just will list targets,

just lint will build the full static site and generate a hyperlink.log file containing broken links.

File: docs/smartmodules/features/operators/array-map.mdx

SmartModule ArrayMaps are used to break apart Records into smaller pieces. This can be very useful for working with your data at a fine granularity. Often, each record in a Topic may actually represent many data points, but we'd like to be able to analyze and manipulate those data points independently. ArrayMap allows us to dig in and break apart these composite records into the smaller units of data that we want to work with.

Let's take a look at an example ArrayMap and walk through how it works and what some sample input and output data might look like.

Generic Example: Transform JSON arrays to records

A common use case is to transform JSON arrays and produce a stream of the records of those arrays. For example, suppose that each element we receive in Fluvio is a JSON array, though we want to interact with the elements of these arrays rather than the arrays themselves. Then, using an ArrayMap, we can transform composite arrays to records that look like this (where this line is a single record):

bash

["a", "b", "c"]

But, we want those elements as a distinct record, like this (where each line is a distinct record):

bash

"a"
"b"
"c"

If you'd like to see a practical example of ArrayMap in action, check out our blog on using ArrayMap to break apart paginated API requests.

Let's dive in and see how to this up in Fluvio.

Create a SmartModule Project

Run smdk generate with the name of the filter and choose the "filter" options:

bash copy="fl"

$ smdk generate array-map
Generating new SmartModule project: array-map
project-group => 'john'
fluvio-smartmodule-cargo-dependency => '"0.3.0"'
🔧   Destination: ~/smdk/array-map ...
🔧   Generating template ...
✔ 🤷   Will your SmartModule use init parameters? · false
✔ 🤷   Which type of SmartModule would you like? · array-map
✔ 🤷   Will your SmartModule be public? · false
Ignoring: /var/folders/5q/jwc86771549058kmbkbqjcdc0000gn/T/.tmp4imt4g/cargo-generate.toml
[1/5]   Done: Cargo.toml
[2/5]   Done: README.md
[3/5]   Done: SmartModule.toml
[4/5]   Done: src/lib.rs
[5/5]   Done: src
🔧   Moving generated files into: `~/smdk/array-map`...
💡   Initializing a fresh Git repository
✨   Done! New project created ~/smdk/array-map

Code Generator for ArrayMap

The code in this generated project takes JSON arrays as input records and returns the elements of those arrays as output records. Let's take a look at the full source, then we'll cover it piece by piece. Let's look at src/lib.rs:

bash copy="fl"

$ cd array-map && cat src/lib.rs

rust

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, RecordData, Result};

#[smartmodule(array_map)]
pub fn array_map(record: &SmartModuleRecord) -> Result<Vec<(Option<RecordData>, RecordData)>> {
    // Deserialize a JSON array with any kind of values inside
    let array: Vec<serde_json::Value> = serde_json::from_slice(record.value.as_ref())?;

    // Convert each JSON value from the array back into a JSON string
    let strings: Vec<String> = array
        .into_iter()
        .map(|value| serde_json::to_string(&value))
        .collect::<core::result::Result<_, _>>()?;

    // Create one record from each JSON string to send
    let records: Vec<(Option<RecordData>, RecordData)> = strings
        .into_iter()
        .map(|s| (None, RecordData::from(s)))
        .collect();
    Ok(records)
}

This ArrayMap essentially has three steps it takes:

Deserialize a JSON array as input and store it in a Vec<Value>
Converts each Value back into a JSON string
Converts each JSON string into a distinct output Record

Let's take this for a test drive and see it in action.

Build the SmartModule

Let's make sure our code compiles. If eveything works as expected, there is a .wasm file generated in the target directory.

bash copy="fl"

$ smdk build
...
Compiling array-map v0.1.0 (~/smdk/array-map)
Finished release-lto [optimized] target(s) in 11.31s

Your SmartModule WASM binary is now ready for use.

Test with SMDK

Let's test our work using the command line test facility.

bash copy="fl"

$ smdk test --text='["a", "b", "c"]'
loading module at: ~/smdk/array-map/target/wasm32-unknown-unknown/release-lto/array_map.wasm
3 records outputed
"a"
"b"
"c"

Great, everything works as expected. Let's test on cluster.

Test on Cluster

Let's create a new Fluvio topic to produce the sample records we want to consume with our SmartModule:

bash copy="fl"

$ fluvio topic create array-map
topic "array-map" created

Now we can produce the sample data to our topic.

bash copy="fl"

$ fluvio produce array-map
> ["a", "b", "c"]
Ok!
> ["d", "e", "f"]
Ok!
> ^C

Let's double check it's all there.

bash copy="fl"

$ fluvio consume array-map -dB
Consuming records from the beginning of topic 'array-map'
["a", "b", "c"]
["d", "e", "f"]

Load SmartModule to Fluvio

The SmartModule can be loaded to local Fluvio Cluster or InfinyOn Cloud, as determined by the [current profile]. In this example, the profile points to InfinyOn Cloud.

bash

$ smdk load
Found SmartModule package: array-map
loading module at: ~/smdk/array-map/target/wasm32-unknown-unknown/release-lto/array_map.wasm
Trying connection to fluvio router.infinyon.cloud:9003
Creating SmartModule: array-map

Rust fluvio smartmodule list to ensure your SmartModule has been uploaded:

bash copy="fl"

$ fluvio smartmodule list
  SMARTMODULE                   SIZE
  john/[email protected]          157.8 KB

SmartModules that have been uploaded on the cluster can be used by other areas of the system (consumers, producers, connectors, etc):

bash copy="fl"

$ fluvio consume array-map -dB --smartmodule=john/[email protected]
Consuming records from the beginning of topic 'array-map'
"a"
"b"
"c"
"d"
"e"
"f"

Congratulations! :tada: Eveything worked as expected!

Publish to SmartModule Hub

It turns out this SmartModule was requested by other data streaming teams in the organization, so we've decided to [publish] it on [SmartModule Hub].

bash copy="fl"

$ smdk publish
Creating package john/[email protected]
.. fill out info in hub/package-meta.yaml
Package hub/array-map-0.1.0.ipkg created
Package uploaded!

Let's double check that the SmartModule is available for download:

bash copy="fl"

$ fluvio hub list
  SMARTMODULE                 Visibility
  john/[email protected]        private
  ...

Congratulations! :tada: Your SmartModule is now available for download in the SmartModule Hub.

File: docs/smartmodules/features/operators/filter.mdx

The simplest type of SmartModule is a filter, which can examine each record in a stream and decide whether to accept or reject it. All accepted records are delivered down the pipeline, and rejected records are discarded. SmartModule applied in consumers or sink connectors filter records after they are stored in a topic, and will not impact persistence - it simply means that records filtered out are not delivered to the consumer. However, SmartModule filters applied to source connectors discard packets before they are stored in the topic and should be used with care.

Getting Practical: Filter Records by JSON fields

In this example, we're going to filter records based on the contents of their JSON fields. Since SmartModules are written using arbitrary Rust code, we can also pull in other crates as dependencies.

Create a SmartModule Project

Run smdk generate with the name of the filter and choose the "filter" options:

bash copy="fl"

$ smdk generate json-filter
project-group => 'john'
Generating new SmartModule project: json-filter
fluvio-smartmodule-cargo-dependency => '"0.2.5"'
🔧   Destination: ~/smdk/json-filter ...
🔧   Generating template ...
🤷   Please set a group name : john
✔ 🤷   Which type of SmartModule would you like? · filter
✔ 🤷   Will your SmartModule use init parameters? · false
✔ 🤷   Will your SmartModule be public? · false
Ignoring: /var/folders/5q/jwc86771549058kmbkbqjcdc0000gn/T/.tmpoM9gda/cargo-generate.toml
[1/5]   Done: Cargo.toml
[2/5]   Done: README.md
[3/5]   Done: SmartModule.toml
[4/5]   Done: src/lib.rs
[5/5]   Done: src
🔧   Moving generated files into: `~/smdk/json-filter`...
💡   Initializing a fresh Git repository
✨   Done! New project created ~/smdk/json-filter

With the SmartModule project created, let's talk about what we will be filtering.

The Data: Server Logs

Suppose we have a web server that accepts HTTP requests from clients, does some
stuff, and then returns a response. It is common for such servers to have an
application logging system where they report various events taking place within the
server so that it may be monitored. We can imagine that this web server is exporting
logs to Fluvio via a producer, and that the logs are formatted as JSON describing
the event that occurred.

For the purposes of this exercise, let's say we have a file that we've stored our logs
into, so that we can manually produce them to a Fluvio topic and consume them back
using our JSON SmartModule. Create a file called server.log with the following
contents:

bash

$ cat server.log
{"level":"info","message":"Server listening on 0.0.0.0:8000"}
{"level":"info","message":"Accepted incoming connection"}
{"level":"debug","message":"Deserializing request from client"}
{"level":"debug","message":"Client request deserialized"}
{"level":"debug","message":"Connecting to database"}
{"level":"warn","message":"Client dropped connection"}
{"level":"info","message":"Accepted incoming connection"}
{"level":"debug","message":"Deserializing request from client"}
{"level":"debug","message":"Client request deserialized"}
{"level":"debug","message":"Connecting to database"}
{"level":"error","message":"Unable to connect to database"}

Each line in this file represents one event that occurred in our server. We can
see that each event is tagged with a "level" describing the significance of the
event, and a "message" with a description about what happened. This style of rating
logs with different levels is a common pattern in application logging, and we're
going to use it as the basis of our filter. Specifically, we're going to write a
filter that excludes all "debug" log, but accepts any "info", "warn", or "error"
logs. In a real-world scenario, this could dramatically help reduce the traffic
and noise in the logs if we were to consume these records into an analytics
platform for inspection.

The Code: Writing our Filter

Let's look at the starter code that we got when we created our Filter template.

bash copy="fl"

$ cd json-filter && cat ./src/lib.rs

rust

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord};

#[smartmodule(filter)]
pub fn filter(record: &SmartModuleRecord) -> Result {
    let string = std::str::from_utf8(record.value.as_ref())?;
    Ok(string.contains('a'))
}

The Record type contains the binary data for a single event in our topic. In our
case, this will be a UTF-8 encoded string that is also a valid JSON value. The
first step we'll need to take is to parse our Record as JSON so that we can
inspect it and determine what level the log is. We can use serde's derive feature
to define types that represents our log data.

rust

#[derive(PartialEq, Eq, PartialOrd, Ord, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
enum LogLevel {
    Debug,
    Info,
    Warn,
    Error
}

#[derive(serde::Deserialize)]
struct StructuredLog {
    level: LogLevel,
    #[serde(rename = "message")]
    _message: String,
}

We're using #[derive(serde::Deserialize)] to implement Deserialize for our types,
which will allow us to convert our raw data into instances of StructuredLog.
We have also defined a LogLevel enum that implements Deserialize as well as
Ord, or "Ordering". When deriving Ord for an enum, the variants may be compared
to one another using < and >, where later-defined variants are "greater than"
earlier-defined variants. In other words, we have LogLevel::Error > LogLevel::Debug
and so on for each pair of LogLevels. Notice also that we have defined a field for
our logs' messages, but it is unused (which is why it is named _message). This
is because our filter will not care about the message in the log, just the level.
However, by including it in our StructuredLog definition, we can be sure that
all logs that we pass through the filter do indeed have a "message" field. In this
way, our filter is also acting as a sort of schema validator, only accepting records
that properly conform to the shape that we expect.

Now, let's write the logic for our filter. We'll start by parsing our raw data into
instances of StructuredLog.

rust

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, Result};

#[smartmodule(filter)]
fn filter(record: &SmartModuleRecord) -> Result {
    let log: StructuredLog = serde_json::from_slice(record.value.as_ref())?;

    todo!()
}

For fluvio_smartmodule::Result, we can bubble-up Results using the ? operator.

Now for the final step, we want our filter to accept all records except for "debug" logs.
In other words, we actually want to keep the records that are "more important" or
"greater than" LogLevel::Debug. Since we have implemented Ord for LogLevel, this
will be a piece of cake! Let's look at all the code for the finished filter.

rust

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, Result};

#[derive(PartialEq, Eq, PartialOrd, Ord, serde::Deserialize)]
#[serde(rename_all = "lowercase")]
enum LogLevel {
    Debug,
    Info,
    Warn,
    Error
}

#[derive(serde::Deserialize)]
struct StructuredLog {
    level: LogLevel,
    #[serde(rename = "message")]
    _message: String,
}

#[smartmodule(filter)]
fn filter(record: &SmartModuleRecord) -> Result {
    let log: StructuredLog = serde_json::from_slice(record.value.as_ref())?;

    // We keep records that are "greater than" debug
    Ok(log.level > LogLevel::Debug)
}

Let's make sure our code compiles. If eveything works as expected, there is a .wasm file generated in the target directory.

bash copy="fl"

$ smdk build
...
Compiling json-filter v0.1.0 (~/smdk/json-filter)
Finished release-lto [optimized] target(s) in 12.78s

Your WASM binary is now ready for use.

Test with SMDK

Now that we've written our filter, let's test using the command line.

Test info log:

bash copy="fl"

$ smdk test --text='{"level":"info","message":"Server listening on 0.0.0.0:8000"}'
loading module at: ~/smdk/json-filter/target/wasm32-unknown-unknown/release-lto/json_filter.wasm
1 records outputted
{"level":"info","message":"Server listening on 0.0.0.0:8000"}

Test debug log:

bash copy="fl"

$ smdk test --text='{"level":"debug","message":"Deserializing request from client"}'
loading module at: ~/smdk/json-filter/target/wasm32-unknown-unknown/release-lto/json_filter.wasm
0 records outputed

Good news! :tada: it works as expected!

Test on Cluster

Let's create a new topic to produce our source data:

bash copy="fl"

$ fluvio topic create server-logs
topic "server-logs" created

Load server.log file to server-logs topic:

bash copy="fl"

$ fluvio produce server-logs -f server.log

Let's double check it's all there.

bash copy="fl"

$ fluvio consume server-logs -B -d
{"level":"info","message":"Server listening on 0.0.0.0:8000"}
{"level":"info","message":"Accepted incoming connection"}
{"level":"debug","message":"Deserializing request from client"}
{"level":"debug","message":"Client request deserialized"}
{"level":"debug","message":"Connecting to database"}
{"level":"warn","message":"Client dropped connection"}
{"level":"info","message":"Accepted incoming connection"}
{"level":"debug","message":"Deserializing request from client"}
{"level":"debug","message":"Client request deserialized"}
{"level":"debug","message":"Connecting to database"}
{"level":"error","message":"Unable to connect to database"}

Load SmartModule to Fluvio

The SmartModule can be loaded to local Fluvio Cluster or InfinyOn Cloud, as determined by the [current profile]. In this example, the profile points to InfinyOn Cloud.

bash copy="fl"

$ smdk load
Loading package at: ~/smdk/json-filter
Found SmartModule package: json-filter
loading module at: ~/smdk/json-filter/target/wasm32-unknown-unknown/release-lto/json_filter.wasm
Trying connection to fluvio router.infinyon.cloud:9003
Creating SmartModule: json-filter

Rust fluvio smartmodule list to ensure your SmartModule has been uploaded:

bash copy="fl"

$ fluvio smartmodule list
  SMARTMODULE                   SIZE
  john/[email protected]        140.1 KB

SmartModules that have been uploaded on the cluster can be used by other areas of the system (consumers, producers, connectors, etc):

bash copy="fl"

$ fluvio consume server-logs -dB --smartmodule=john/[email protected]
Consuming records from the beginning of topic 'server-logs'
{"level":"info","message":"Server listening on 0.0.0.0:8000"}
{"level":"info","message":"Accepted incoming connection"}
{"level":"warn","message":"Client dropped connection"}
{"level":"info","message":"Accepted incoming connection"}
{"level":"error","message":"Unable to connect to database"}

Congratulations! :tada: Everything worked as expected!

Publish to SmartModule Hub

It turns out this SmartModule was requested by other data streaming teams in the organization, so we've decided to [publish] it on [SmartModule Hub].

bash copy="fl"

$ smdk publish
Creating package john/[email protected]
.. fill out info in hub/package-meta.yaml
Package hub/json-filter-0.1.0.ipkg created
Package uploaded!

Let's double check that the SmartModule is available for download:

bash copy="fl"

$ fluvio hub list
  SMARTMODULE                 Visibility
  john/[email protected]      private
  ...

Congratulations! :tada: Your SmartModule is now available for download in the SmartModule Hub.

File: docs/smartmodules/features/operators/filter-map.mdx

SmartModule FilterMaps are used to both transform and potentially filter records from a stream at the same time. This can be useful for a number of
scenarios including working with data with nullable fields, or working with subsets of event data. In these cases, FilterMap allows us discard irrelevant data - such as records with null fields or event types that we don't care about - while also performing meaningful work with relevant data - such as reformatting fields we've extracted or events we've gathered.

FilterMap functions work by returning an Option of a new record. To discard a record from the stream, return None. Otherwise, transform
the record according to your needs and return it as Some(record).

Let's dive in and see how to use this in practice. You can find the full code for this doc's example in the fluvio-smartmodule-examples repository.

Getting Practical: Transform `Order Ready` events in a mobile shopping App

For this example, we'll write a simplified version of the online grocery-notifications example we blogged about. In this simplified example, we'll consider a stream that has two event types: Account Created and Order Ready. We'll use FilterMap to discard account_created events, and to transform order_ready events into text messages that should be sent to the user.

Let's take a look at some sample input for this example:

bash

{"type":"account_created","account_id":"1","username":"bill9876","preferred_name":"Bill","phone_number":"1-800-234-5678"}
{"type":"order_ready","account_id":"1","sms_number":"1-800-234-5678","sms_name":"Bill"}
{"type":"account_created","account_id":"2","username":"mary","preferred_name":"Mary","phone_number":"1-222-222-2222"}
{"type":"order_ready","account_id":"2","sms_number":"1-222-222-2222","sms_name":"Mary"}

We'd like to discard the account_created events, keep the order_ready events, and
transform the order_ready events to a nice message that should be sent to the user.
The output records in this stream should look like this:

bash

{"number":"1-800-234-5678","message":"Hello Bill, your groceries have been collected and are ready to pick up!"}
{"number":"1-222-222-2222","message":"Hello Mary, your groceries have been collected and are ready to pick up!"}

Create a SmartModule Project

Run smdk generate with the name of the filter and choose the "filter" options:

bash copy="fl"

$ smdk generate filter-map
project-group => 'john'
fluvio-smartmodule-cargo-dependency => '"0.3.0"'
🔧   Destination: ~/smdk/filter-map ...
🔧   Generating template ...
✔ 🤷   Will your SmartModule use init parameters? · false
✔ 🤷   Which type of SmartModule would you like? · filter-map
✔ 🤷   Will your SmartModule be public? · false
Ignoring: /var/folders/5q/jwc86771549058kmbkbqjcdc0000gn/T/.tmp5Yi3lU/cargo-generate.toml
[1/5]   Done: Cargo.toml
[2/5]   Done: README.md
[3/5]   Done: SmartModule.toml
[4/5]   Done: src/lib.rs
[5/5]   Done: src
🔧   Moving generated files into: `~/smdk/filter-map`...
💡   Initializing a fresh Git repository
✨   Done! New project created ~/smdk/filter-map

The generator created a sample code for us, let's go ahead and update it.

The Code: Writing our FilterMap

We'll want to cd into the project directory for the rest of the commands to work:

bash copy="fl"

$ cd filter-map

Now, let's jump right into the code. Copy and paste the following block into the src/lib.rs file in the new SmartModule project:

rust

use fluvio_smartmodule::{smartmodule, SmartModuleRecord, RecordData, Result};
use serde::{Deserialize, Serialize};

/// Events that may take place in an online grocery service
#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum GroceryEvent {
    AccountCreated {
        account_id: String,
        username: String,
        preferred_name: String,
        phone_number: String,
    },
    OrderReady {
        account_id: String,
        sms_number: String,
        sms_name: String,
    },
}

#[smartmodule(filter_map)]
fn filter_map(record: &SmartModuleRecord) -> Result<Option<(Option<RecordData>, RecordData)>> {
    let event: GroceryEvent = match serde_json::from_slice(record.value.as_ref()) {
        Ok(event) => event,
        Err(_) => return Ok(None), // Skip if we fail to parse JSON
    };

    let sms_message = match event {
        GroceryEvent::OrderReady {
            sms_name,
            sms_number,
            ..
        } => serde_json::json!({
            "number": sms_number,
            "message": format!(
                "Hello {}, your groceries have been collected and are ready to pick up!",
                sms_name
            ),
        }),
        _ => return Ok(None),
    };

    let message_json = serde_json::to_string(&sms_message)?;
    Ok(Some((record.key.clone(), message_json.into())))
}

Let's break down what's happening here. First, we have a GroceryEvent enum which represents the different types of input record we are expecting. We're using serde to automatically serialize and deserialize our JSON into this enum type.

Next, we check what type of event we received. If it's an order_ready event, we transform it, picking the phone number and account preferred name and creating a record that contains a text-friendly message that should be sent to the user. If it is any other event type, we filter it out by returning Ok(None).

Build the SmartModule

Let's make sure our code compiles. If everything works as expected, there is a .wasm file generated in the target directory.

bash copy="fl"

$ smdk build
...
Compiling filter-map v0.1.0 (~/smdk/filter-map)
Finished release-lto [optimized] target(s) in 11.97s

Your SmartModule WASM binary is now ready for use.

Test with SMDK

Let's test our work using the command line test facility.

Test filter component to ensure non-matching records are eliminated:

bash copy="fl"

$ smdk test --text='{"type":"account_created","account_id":"1","username":"bill9876","preferred_name":"Bill","phone_number":"1-800-234-5678"}'
loading module at: ~/smdk/filter-map/target/wasm32-unknown-unknown/release-lto/filter_map.wasm
0 records outputted

Test map part to ensure matching records are transformed:

bash copy="fl"

$ smdk test --text='{"type":"order_ready","account_id":"1","sms_number":"1-800-234-5678","sms_name":"Bill"}'
loading module at: ~/smdk/filter-map/target/wasm32-unknown-unknown/release-lto/filter_map.wasm
1 records outputted
{"message":"Hello Bill, your groceries have been collected and are ready to pick up!","number":"1-800-234-5678"}

Test on Cluster

Let's create a new Fluvio topic to produce the sample records we want to consume with our SmartModule:

bash copy="fl"

$ fluvio topic create filter-map
topic "filter-map" created

Now, let's put our sample data into a file and produce it to our topic.
Create groceries.txt with the following contents:

bash

{"type":"account_created","account_id":"1","username":"bill9876","preferred_name":"Bill","phone_number":"1-800-234-5678"}
{"type":"order_ready","account_id":"1","sms_number":"1-800-234-5678","sms_name":"Bill"}
{"type":"account_created","account_id":"2","username":"mary","preferred_name":"Mary","phone_number":"1-222-222-2222"}
{"type":"order_ready","account_id":"2","sms_number":"1-222-222-2222","sms_name":"Mary"}

Now we can produce the sample data to our topic.

bash copy="fl"

$ fluvio produce filter-map -f ./groceries.txt

Let's double check it's all there.

bash copy="fl"

$ fluvio consume filter-map -dB
Consuming records from the beginning of topic 'filter-map'
{"type":"account_created","account_id":"1","username":"bill9876","preferred_name":"Bill","phone_number":"1-800-234-5678"}
{"type":"order_ready","account_id":"1","sms_number":"1-800-234-5678","sms_name":"Bill"}
{"type":"account_created","account_id":"2","username":"mary","preferred_name":"Mary","phone_number":"1-222-222-2222"}
{"type":"order_ready","account_id":"2","sms_number":"1-222-222-2222","sms_name":"Mary"}

Load SmartModule to Fluvio

The SmartModule can be loaded to local Fluvio Cluster or InfinyOn Cloud, as determined by the [current profile]. In this example, the profile points to InfinyOn Cloud.

bash copy="fl"

$ smdk load
Loading package at: ~/smdk/filter-map
Found SmartModule package: filter-map
loading module at: ~/smdk/filter-map/target/wasm32-unknown-unknown/release-lto/filter_map.wasm
Trying connection to fluvio router.infinyon.cloud:9003
Creating SmartModule: filter-map

Rust fluvio smartmodule list to ensure your SmartModule has been uploaded:

bash copy="fl"

$ fluvio smartmodule list
  SMARTMODULE                   SIZE
  john/[email protected]         166.6 KB

SmartModules that have been uploaded on the cluster can be used by other areas of the system (consumers, producers, connectors, etc):

bash copy="fl"

$ fluvio consume filter-map -dB --smartmodule=john/[email protected]
Consuming records from the beginning of topic 'filter-map'
{"message":"Hello Bill, your groceries have been collected and are ready to pick up!","number":"1-800-234-5678"}
{"message":"Hello Mary, your groceries have been collected and are ready to pick up!","number":"1-222-222-2222"}

Congratulations! :tada: Everything worked as expected!

Publish to SmartModule Hub

Let's [publish] this SmartModule to [SmartModule Hub] to make accessible to others.

bash copy="fl"

$ smdk publish
Creating package john/[email protected]
.. fill out info in hub/package-meta.yaml
Package hub/filter-map-0.1.0.ipkg created
Package uploaded!

Let's double check that the SmartModule is available for download:

bash copy="fl"

$ fluvio hub list
  SMARTMODULE                 Visibility
  john/[email protected]        private
  ...

Congratulations! :tada: Your SmartModule is now available for download in the SmartModule Hub.

File: docs/smartmodules/features/operators/map.mdx

SmartModule Maps are used to transform or edit each Record in a stream. We say that these SmartModules "map" each input record into a new output record by applying a function to the input data. This type of SmartModule may be used for many use-cases, such as:

Narrowing large records into a smaller subset of important fields
Scrubbing sensitive fields of data to be invisible to downstream consumers
Computing rich, derived fields from simple raw data

Let's create a brand-new SmartModule Map to see what a minimal working
example looks like.

Create a SmartModule Project

Run smdk generate with the name of the map and choose the "map" options:

bash copy="fl"

$ smdk generate map-example
Generating new SmartModule project: map-example
project-group => 'john'
fluvio-smartmodule-cargo-dependency => '"0.2.5"'
🔧   Destination: ~/smdk/map-example ...
🔧   Generating template ...
✔ 🤷   Will your SmartModule use init parameters? · false
✔ 🤷   Which type of SmartModule would you like? · map
✔ 🤷   Will your SmartModule be public? · false
Ignoring: /var/folders/5q/jwc86771549058kmbkbqjcdc0000gn/T/.tmpNFObJj/cargo-generate.toml
[1/5]   Done: Cargo.toml
[2/5]   Done: README.md
[3/5]   Done: SmartModule.toml
[4/5]   Done: src/lib.rs
[5/5]   Done: src
🔧   Moving generated files into: `~/smdk/map-example`...
💡   Initializing a fresh Git repository
✨   Done! New project created ~/smdk/map-example

We should see a new folder has been created for our project map-example. Let's navigate inside and take a look at the sample Map generated for us by the template:

bash copy="fl"

$ cd map-example && cat ./src/lib.rs

rust

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord, RecordData};

#[smartmodule(map)]
pub fn map(record: &SmartModuleRecord) -> Result<(Option<RecordData>, RecordData)> {
    let key = record.key.clone();

    let string = std::str::from_utf8(record.value.as_ref())?;
    let int = string.parse::()?;
    let value = (int * 2).to_string();

    Ok((key, value.into()))
}

Let's break down what's happening here:

Firstly, #[smartmodule(map)] marks the entry point for this SmartModule Map.
There may only be one of these in the project, and it is called once for each
record in the data stream.
The annotated function fn map may be named anything, but it must take a
single &Record argument. This variable contains the contents of one record
in the stream, and you may read the Key and Value of this record as bytes.
The fn map function must return a new Key and Value for the output record.
The Key is the Option<RecordData> and the Value is the RecordData in the
return type. RecordData is a helper type that may be constructed from any
type that has impl Into<Vec> such as String, by using .into().
At any point in the SmartModule, errors may be returned using ? or via
Err(e.into()). This works for any error type that has impl std::error::Error.

This template SmartModule will parse each record as an i32 integer, then multiply that value by 2.

Let's make sure our code compiles. If everything works as expected, there will be a .wasm file generated in the target directory.

bash copy="fl"

$ smdk build
...
Compiling map-example v0.1.0 (~/smdk/map-example)
Finished release-lto [optimized] target(s) in1 12.83s

Your WASM binary is now ready for use.

Test with SMDK

Now that we've written our map, let's test using the command line.

bash copy="fl"

$ smdk test --text=6
loading module at: ~/smdk/map-example/target/wasm32-unknown-unknown/release-lto/map_example.wasm
1 records outputted
12

Good news! :tada: it works as expected!

Test on Cluster

Let's create a new topic to produce our source data:

bash copy="fl"

$ fluvio topic create map-double
topic "map-double" created

In a new terminal, produce the following entries:

bash copy="fl"

$ fluvio produce map-double
> 1
Ok!
> 2
Ok!
> 3
Ok!
> 4
Ok!
> 5
Ok!
> ^C

Let's double check it's all there.

bash copy="fl"

$ fluvio consume map-double -B -d
Consuming records from the beginning of topic 'map-double'
1
2
3
4
5

Load SmartModule to Fluvio

The SmartModule can be loaded to local Fluvio Cluster or InfinyOn Cloud, as determined by the [current profile]. In this example, the profile points to InfinyOn Cloud.

bash copy="fl"

$ smdk load
Loading package at: ~/smdk/map-example
Found SmartModule package: map-example
loading module at: ~/smdk/map-example/target/wasm32-unknown-unknown/release-lto/map_example.wasm
Trying connection to fluvio router.infinyon.cloud:9003
Creating SmartModule: map-example

Rust fluvio smartmodule list to ensure your SmartModule has been uploaded:

bash copy="fl"

$ fluvio smartmodule list
 SMARTMODULE                    SIZE
 john/[email protected]         87.5 KB

SmartModules that have been uploaded on the cluster can be used by other areas of the system (consumers, producers, connectors, etc):

bash copy="fl"

$ fluvio consume map-double -B --smartmodule=john/[email protected]
Consuming records from the beginning of topic 'map-double'
2
4
6
8
10

Congratulations! :tada: Everything worked as expected!

Publish to SmartModule Hub

As bonus, let's [publish] this SmartModule to [SmartModule Hub].

bash copy="fl"

$ smdk publish
Creating package john/[email protected]
.. fill out info in hub/package-meta.yaml
Package hub/map-example-0.1.0.ipkg created
Package uploaded!

Let's double check that the SmartModule is available for download:

bash copy="fl"

$ fluvio hub list
  SMARTMODULE                 Visibility
  john/[email protected]      private
  ...

Congratulations! :tada: Your SmartModule is now available for download in the SmartModule Hub.

File: docs/smartmodules/features/chaining.mdx

SmartModule Chaining

A group of SmartModules forms a transformation chain - sequential invocations of each SmartModule according to the order defined upon creation. The output of the first invocation becomes the input of the second, the output of the second becomes the input of the third, and so on.

Both Fluvio Producer and Fluvio Consumer support Transformation Chaining. However, the actual code execution happens in different places. For Producer, the transformation
happens before the data is sent to the topic on the SPU, inside the Producer's process, hence utilizing the resources of the client. For Consumer, the transformation takes place on SPU before the data is sent
to the Consumers.

Transformation Chaining is available on the following components:

Fluvio Client
Fluvio CLI
Connectors

Configuration

As we previously noted, each Transformation is a SmartModule. More precisely, it is a SmartModule plus some custom parameters.
Most often, Transformation Chaining is configured in yaml config file.

Example:

yaml

transforms:
  - uses: infinyon/[email protected]
    with:
      spec:
        - operation: default
          spec:
            source: "http"

we have one transformation, which is a SmartModule named infinyon/[email protected].
The name must match a SmartModule previously downloaded to the Cluster:

bash

fluvio sm list
  SMARTMODULE              SIZE
  infinyon/[email protected]      608.4 KB

Everything under with section in the config is treated as key/value pairs and passed to SmartModule as init parameters. The value in that pair is a JSON-serialized string.
In our example, the SmartModule will receive one pair:

json

"spec" : "[{\"operation\":\"default\",\"spec\":{\"source\":\"http\"}}]"

This is an example of how this particular SmartModule process parameters (the code is just for demonstration and may differ from real implementation):

rust

#[smartmodule(init)]
fn init(params: SmartModuleExtraParams) -> Result<()> {
    if let Some(raw_spec) = params.get("spec") {
        match serde_json::from_str(raw_spec) {
            Ok(spec) => {
                SPEC.set(spec).expect("spec is already initialized");
                Ok(())
            }
            Err(err) => {
                eprintln!("unable to parse spec from params: {:?}", err);
                Err(eyre::Report::msg(
                    "could not parse the specification from `spec` param",
                ))
            }
        }
    } else {
        Err(SmartModuleInitError::MissingParam("spec".to_string()).into())
    }
}

The value in parameters can be strings, maps, or sequences. In this pseudo example, all values are valid:

yaml

transforms:
  - uses: mygroup/[email protected]
    with:
      map_param_name:
        key1: value1
        key2:
          nested: "value2"
      seq_param_name: ["value1", "value2"]
      string_param_name: "value"

Using Multiple SmartModules

Multiple SmartModules can be activated in series. The output of the previous SmartModule is used as the input of the next SmartModule. Therefore the ordering of the SmartModules in the chain is important.

yaml

transforms:
  - uses: infinyon/[email protected]
    with:
      spec:
        - operation: shift
          spec:
            fact: "animal.fact"
            length: "length"
  - uses: infinyon/[email protected]
    with:
      regex: "[Cc]at"

In this example, the jolt transformation is performed first, then its output is the input the regex-filter.

File: docs/smartmodules/features/deduplication.mdx

The Deduplication feature in Fluvio is the way to deduplicate records based on their keys.

Deduplication can be used on a topic.
It will drop the duplicate records in a window which is defined by the configured bounds.

The current supported bounds are age and count, they are explained in the bounds section.

Behavior

The deduplication function is deterministic and survives restarts. During restart the deduplication algorithm scans the data stream and rebuilds the memory object.

Example topic config

Example configuration on topic:

yaml

# topic.yaml
version: 0.1.0
meta:
  name: topic-with-dedup
deduplication:
  bounds:
    count: 5 # remember at least 5 last records
    age: 5s # remember records for at least 5 seconds
  filter:
    transform:
      uses: infinyon-labs/[email protected]

A topic can be created using this config file like so:

bash copy="fl"

$ fluvio topic create -c topic.yaml

After creating the topic, it can be tested like so:

bash copy="fl"

$ fluvio produce topic-with-dedup --key-separator :
1:2
1:2
2:5

bash copy="fl"

$ fluvio consume -B topic-with-dedup
2
5

Bounds

Parameter	default	type	optional	description
count	-	Integer	false	Minimum number of records the filter will remember. It doesn't guarantee to remember records that came `count` records before now.
age	-	Integer	true	Minimum amount of time this filter will remember a record for. It can be specified using this format: `15days 2min 2s`, or `2min 5s`, or `15ms`

Implementation

The specific algorithm used for deduplication is defined by a smartmodule. Right now the only smartmodule for this is the dedup-filter smartmodule.

The dedup-filter splits the data into smaller chunks and keeps these chunks in memory. Each chunk has an age so it knows if it can
drop a chunk based on it's age.

When count of total records in memory exceeds the configured bounds.count, it will check if it can delete the oldest record chunk using the configured bounds.age and delete it if it can. This allows for fast deletion of old data with minimum amount of bookkeeping.

The chunking of data introduces a small memory usage overhead but allows the filter to have no spikes in execution time and memory usage.

File: docs/smartmodules/features/lookback.mdx

The Lookback feature in Fluvio SmartModule is the way to access records from the data stream before the SmartModule starts allowing it to build its internal state depending on what the topic currently has.

If configured, the Lookback phase is guaranteed to take place after init method but before the first record is processed.

To activate Lookback SmartModule must fit the following criteria:

A function annotated with #[smartmodule(look_back)] macro is present in the code. This function is used to pass requested records to SmartModule:

rust

#[smartmodule(look_back)]
pub fn look_back(record: &SmartModuleRecord) -> Result<()> {
    ...
}

lookback parameter is specified in transform configuration. It defines a set of lookback records that SmartModule receives. Supported ways to define the set:
- by size (last N records existing in the topic)
- by age (records that are younger than the specified age)
- by size and age (max N records younger than the specified age)

yaml

transforms:
  - uses: local/[email protected]
    lookback:
      age: 30m # we want only records that are younger than 30 minutes
      last: 10 # we want maximum of 10 records (last)

If Fluvio topic is empty, look_back is never called.

-> If you start processing from an offset other than the end, you will receive records both as lookback record and as regular processing

-> Lookback is only supported for SmartModules that are run on SPU. This implies that Source Connectors don't support it.

Monotonically increasing values

This is an example of SmartModule that leverages Lookback functionality. It reads the last record from a topic and
only allows a record that is higher than the previous value.

rust

use std::sync::atomic::{AtomicI32, Ordering::SeqCst};
use fluvio_smartmodule::{smartmodule, SmartModuleRecord, Result};

static PREV: AtomicI32 = AtomicI32::new(0);

#[smartmodule(filter)]
pub fn filter(record: &SmartModuleRecord) -> Result {
   let string = std::str::from_utf8(record.value.as_ref())?;
   let current: i32 = string.parse()?;
   let last = PREV.load(SeqCst);
   if current > last {
       PREV.store(current, SeqCst);
       Ok(true)
   } else {
       Ok(false)
   }
}

#[smartmodule(look_back)]
pub fn look_back(record: &SmartModuleRecord) -> Result<()> {
   let string = std::str::from_utf8(record.value.as_ref())?;
   let last: i32 = string.parse()?;
   PREV.store(last, SeqCst);
   Ok(())
}

Using this transforms.yaml config we define that we only need one last record:

yaml title="transforms.yaml"

transforms:
 - uses: local/[email protected]
   lookback:
     last: 1

Test with Fluvio Cli

First, let's generate a new SmartModule using SMDK tool:

bash copy="fl"

$ smdk generate
🤷   Project Name: filter-with-lookback
🤷   Please set a group name: local
✔ 🤷   Which type of SmartModule would you like? · filter
✔ 🤷   Will your SmartModule use init parameters? · false
✔ 🤷   Will your SmartModule be public? · false
[1/7]   Done: .gitignore
[2/7]   Done: Cargo.toml
[3/7]   Done: README.md
[4/7]   Done: SmartModule.toml                                                                    
[5/7]   Done: rust-toolchain.toml                                                                 [6/7]   Done: src/lib.rs                     
[7/7]   Done: src

Then, put the code snippet from above into src/lib.rs file.

Now we are ready to build and load SmartModule to the cluster:

bash copy="fl"

$ smdk build
$ smdk load

Create a topic

bash copy="fl"

$ smdk test-topic

Let's produce 3 records into our topic:

bash copy="fl"

$ fluvio produce test-topic
> 1
Ok!
> 2
Ok!
> 3
Ok!

In another terminal, run Fluvio Consumer with transforms.yaml file from the example above. It will use our SmartModule with configured lookback:

bash copy="fl"

$ fluvio consume test-topic  --transforms-file transforms.yaml

If you insert another record in Producer, Consumer will only output it only if it's greater than 3 (last record value when Consumer started).

Test with SMDK tool

We will use smdk test subcommand to verify that our SmartModule works.

By --lookback-last 1 we specify lookback parameter to "read last 1 record".
By --record "N" we specify records that will exist before the SmartModule start processing.

The following commands will pass records "2" and "3" to look_back and record "4" to filter function:

bash copy="fl"

$ smdk test --text 4 --lookback-last 1 --record 2 --record 3
4

The output result is 4. But if we run:

bash copy="fl"

$ smdk test --text 1 --lookback-last 1 --record 2 --record 3

the record will be filtered out as expected.

File: docs/smartmodules/developers/build.mdx

This section covers building and testing SmartModules locally, and the next section describes how to load them on the Cluster for live traffic testing.

Build - Operation

Navigate to my-filter project directory and run build:

bash copy="fl"

$ smdk build
...
Compiling my-filter v0.1.0 (~/smdk/my-filter)
Finished release-lto [optimized] target(s) in 12.65s

The build process generated an binary optimized for WebAssembly. We are now ready to test it.

Test - Operation

When testing a SmartModule you can provide the test input via direct text, specifying a file,
or reading from the stdin.

Let's perform the matching test for our my-filter that looks for letter a:

bash copy="fl"

$ smdk test --text cat
loading module at: ~/smdk/my-filter/target/wasm32-unknown-unknown/release-lto/my_filter.wasm
1 records outputted
cat

Next, a non-matching test, reading the input from the stdin:

bash copy="fl"

$ echo -n dog | smdk test --stdin
loading module at: ~/smdk/my-filter/target/wasm32-unknown-unknown/release-lto/my_filter.wasm
0 records outputted

Checkout the smdk test -h for additional parameters.

Checkout the next section on how upload the SmartModule to your local cluster or InfinyOn Cloud for live traffic testing.

Steps

File: docs/smartmodules/developers/generate.mdx

SMDK generate helps developers build a sample SmartModule project by answering a few simple questions.

Generate - Operation

SMDK generate commands runs a wizard and builds a sample project in a subdirectory

bash copy="fl"

$ smdk generate my-filter

Generating new SmartModule project: my-filter
fluvio-smartmodule-cargo-dependency => '"0.2.5"'
🔧   Destination: ~/smdk/my-filter ...
🔧   Generating template ...
🤷   Will your SmartModule use init parameters? · false
🤷   Which type of SmartModule would you like? · filter
🤷   Will your SmartModule be public? · false
🤷   Please set a group name : acme
[1/5]   Done: Cargo.toml
[2/5]   Done: README.md
[3/5]   Done: SmartModule.toml
[4/5]   Done: src/lib.rs
[5/5]   Done: src
🔧   Moving generated files into: `~/smdk/my-filter`...
💡   Initializing a fresh Git repository
✨   Done! New project created ~/smdk/my-filter
hub: hubid acme is set

The generator created Rust project ready to compile:

bash

$ tree
.
├── Cargo.toml
├── README.md
├── SmartModule.toml
└── src
    └── lib.rs

This is a simple SmartModule filter matching for all data records that contains letter a:

bash

use fluvio_smartmodule::{smartmodule, Result, SmartModuleRecord};

#[smartmodule(filter)]
pub fn filter(record: &SmartModuleRecord) -> Result {
    let string = std::str::from_utf8(record.value.as_ref())?;
    Ok(string.contains('a'))
}

Note the SmartModule.toml file. This file contains SmartModule parameters required to load the file in the Cluster and publish it to SmartModule Hub.

bash copy="fl"

$ cat SmartModule.toml

[package]
name = "my-filter"
group = "acme"
version = "0.1.0"
apiVersion = "0.1.0"
description = ""
license = "Apache-2.0"

[[params]]
name = "input"
description = "input description"

Sections

package is used to build the SmartModule FQDN acme/[email protected], and the description to publish to SmartModule Hub. The group name is equivalent to the package owner in the Hub.
params defines the command line parameters by the SmartModule internal logic.

The project is ready to build and test. Checkout the next section for instructions.

Docker & Self-Hosted Deployment Verified OCI

1. Project Overview & Quickstart (fluvio-community/fluvio) Copy Chapter

File: README.md Copy Chapter

Quick Start - Get started with Fluvio in 2 minutes or less! Copy Chapter

Step 1. Download Fluvio Version Manager: Copy Section

Step 2. Start a cluster: Copy Section

Step 3. Create Topic: Copy Section

Step 4. Produce to Topic, Consume From Topic: Copy Section

Using Pre-Built Fluvio Versions Copy Chapter

Check Fluvio Core Documentation

Check Stateful DataFlow Documentation

Learn how to build custom connectors

Learn how to build custom smart modules

Clients Copy Section

Contributing Copy Chapter

Community Copy Chapter

Projects Using Fluvio Copy Section

Community Connectors Copy Section

Community Development Resources Copy Chapter

Contributors are awesome Copy Section

Community Copy Chapter

Projects Using Fluvio Copy Section

License Copy Chapter

File: connector/docker/README.md Copy Chapter

Introduction

To build image

To run image

File: crates/fluvio-storage/README.md Copy Chapter

Building storage cli

Running storage cli

File: crates/fluvio-spu-schema/README.md Copy Chapter

flv-api-spu

License Copy Chapter

Contribution Copy Section

File: crates/fluvio-spu/README.md Copy Chapter

Running SPU in development mode Copy Chapter

File: crates/fluvio-socket/README.md Copy Chapter

fluvio-protocol

build fluvio send utility

License Copy Chapter

Contribution Copy Section

File: crates/fluvio-smartmodule/README.md Copy Chapter

Fluvio SmartModules

Writing SmartModules Copy Chapter

Filtering Copy Section

Mapping Copy Section

Aggregate Copy Section

ArrayMap Copy Section

Init Copy Section

License Copy Chapter

Contribution Copy Section

2. Official Technical Reference & Guides (fluvio-community/fluvio-docs) Copy Chapter

File: README.md Copy Chapter

Contributing Copy Chapter

Prerequisites Copy Section

Installation Copy Section

Development Copy Section

Create Versions Copy Section

Release and Publish Copy Chapter

Environments Copy Section

Dev tooling Copy Chapter

File: docs/smartmodules/features/operators/array-map.mdx Copy Chapter

sidebar_position: 40 title: "Array Map" Copy Chapter

Generic Example: Transform JSON arrays to records Copy Chapter

Create a SmartModule Project Copy Section

Code Generator for ArrayMap Copy Section

Build the SmartModule Copy Section

Test with SMDK Copy Section

Test on Cluster Copy Section

Load SmartModule to Fluvio Copy Section

Publish to SmartModule Hub Copy Chapter

Read next Copy Chapter

File: docs/smartmodules/features/operators/filter.mdx Copy Chapter

sidebar_position: 10 title: "Filter" Copy Chapter

Getting Practical: Filter Records by JSON fields Copy Chapter

Create a SmartModule Project Copy Section

The Data: Server Logs Copy Section

The Code: Writing our Filter Copy Section

Test with SMDK Copy Section

Test on Cluster Copy Section

1. Project Overview & Quickstart (fluvio-community/fluvio)

File: README.md

Quick Start - Get started with Fluvio in 2 minutes or less!

Step 1. Download Fluvio Version Manager:

Step 2. Start a cluster:

Step 3. Create Topic:

Step 4. Produce to Topic, Consume From Topic:

Using Pre-Built Fluvio Versions

Clients

Contributing

Community

Projects Using Fluvio

Community Connectors

Community Development Resources

Contributors are awesome

Community

Projects Using Fluvio

License

File: connector/docker/README.md

File: crates/fluvio-storage/README.md

File: crates/fluvio-spu-schema/README.md

License

Contribution

File: crates/fluvio-spu/README.md

Running SPU in development mode

File: crates/fluvio-socket/README.md

License

Contribution

File: crates/fluvio-smartmodule/README.md

Writing SmartModules

Filtering

Mapping

Aggregate

ArrayMap

Init

License

Contribution

2. Official Technical Reference & Guides (fluvio-community/fluvio-docs)

File: README.md

Contributing

Prerequisites

Installation

Development

Create Versions

Release and Publish

Environments

Dev tooling

File: docs/smartmodules/features/operators/array-map.mdx

sidebar_position: 40 title: "Array Map"

Generic Example: Transform JSON arrays to records

Create a SmartModule Project

Code Generator for ArrayMap

Build the SmartModule

Test with SMDK

Test on Cluster

Load SmartModule to Fluvio

Publish to SmartModule Hub

Read next

File: docs/smartmodules/features/operators/filter.mdx

sidebar_position: 10 title: "Filter"

Getting Practical: Filter Records by JSON fields

Create a SmartModule Project

The Data: Server Logs

The Code: Writing our Filter

Test with SMDK

Test on Cluster

Publish to SmartModule Hub

Read next

File: docs/smartmodules/features/operators/filter-map.mdx

sidebar_position: 20 title: "Filter Map"

Getting Practical: Transform `Order Ready` events in a mobile shopping App

Create a SmartModule Project

The Code: Writing our FilterMap

Build the SmartModule

Test with SMDK

Test on Cluster

Publish to SmartModule Hub

Read next

File: docs/smartmodules/features/operators/map.mdx

sidebar_position: 30 title: "Map"