Prerequisites

This guide uses local Fluvio cluster. If you need to install it, please follow the instructions at here.

Introduction

The Filter Operator is a powerful tool that allows you to filter messages based on a condition. It is a simple operator that takes a single input and returns a boolean value. If the input meets the condition, the operator will return true. Otherwise, it will return false.

Syntax

Here is a simple filter operator. It takes a string as input and returns a boolean value. If the input string contains a question mark, the operator will return true. Otherwise, it will return false.

    transforms:
      - operator: filter
        run: |
          fn filter_questions(input: String) -> Result<bool> {
            Ok(input.contains("?"))
          }

Running the Example

Coyp and paste following config and save it as dataflow.yaml.

apiVersion: 0.5.0
meta:
  name: filter-example
  version: 0.1.0
  namespace: examples

config:
  converter: raw

topics:
  sentences:
    schema:
      value:
        type: string

  questions:
    schema:
      value:
        type: string

services:
  filter-service:
    sources:
      - type: topic
        id: sentences

    transforms:
      - operator: filter
        run: |
          fn filter_questions(input: String) -> Result<bool> {
            Ok(input.contains("?"))
          }

    sinks:
      - type: topic
        id: questions

To run example:

$ sdf run --ephemeral

Produce sentences to in sentence topic:

$ fluvio produce sentence

Consume topic questions to retrieve the result in another terminal:

$ fluvio consume questions

Product the following sentences:

Hello world
Are you there?

You should only see message that contains a question mark:

Are you there?

Conclusion

We just covered one of the most basic operators in SDF, the Filter Operator. Combining with other basic operators, you can build more complex dataflows to process your data.

Reference