project-flogo / stream

Elegant stream processing pipeline written entirely in Golang

Streams is a pipeline based, stream processing action for the Project Flogo Ecosystem

Flogo Stream

Edge devices have the potential for producing millions or even billions of events at rapid intervals, often times the events on their own are meaningless, hence the need to provide basic streaming operations against the slew of events.

A native streaming action as part of the Project Flogo Ecosystem accomplishes the following primary objectives:

• Enables apps to implement basic streaming constructs in a simple pipeline fashion
• Provides non-persistent state for streaming operations
  • Streams are persisted in memory until the end of the pipeline
• Serves as a pre-process pipeline for raw data to perform basic mathematical and logical operations. Ideal for feeding ML models

Some of the key highlights include:

😀 Simple pipeline construct enables a clean, easy way of dealing with streams of data
⏳ Stream aggregation across streams using time or event tumbling & sliding windows
🙌 Join streams from multiple event sources
🌪 Filter out the noise with stream filtering capabilities

Getting Started

We’ve made building powerful streaming pipelines as easy as possible. Develop your pipelines using:

• A simple, clean JSON-based DSL
• Golang API

See the sample below of an aggregation pipeline (for brevity, the triggers and metadata of the resource has been omitted). Also don’t forget to check out the examples in the repo.

  "stages": [
    {
      "ref": "github.com/project-flogo/stream/activity/aggregate",
      "settings": {
        "function": "sum",
        "windowType": "timeTumbling",
        "windowSize": "5000"
      },
      "input": {
        "value": "=$.input"
      }
    },
    {
      "ref": "github.com/project-flogo/contrib/activity/log",
      "input": {
        "message": "=$.result"
      }
    }
  ]

Try out the example

Firstly you should install the install the Flogo CLI.

Next you should download our aggregation example agg-flogo.json.

We'll create a our application using the example file, we'll call it myApp

$ flogo create -f agg-flogo.json myApp

Now, build it...

$ cd myApp/
$ flogo build

Activities

Flogo Stream also provides some activities to assist in stream processing.

• Aggregate : This activity allows you to aggregate data and calculate an average or sliding average.
• Filter : This activity allows you to filter out data in a streaming pipeline.

License

Flogo source code in this repository is under a BSD-style license, refer to LICENSE

stream's Issues

Examples for using stream api

Can we have some examples on how to use all the features of this project in go code?

Unable to build from examples shipped

Current behavior (how does the issue manifest):
Running flogo-cli to build example fails with error

Expected behavior:
Build should generate app
Minimal steps to reproduce the problem (not required if feature enhancement):

$ flogo create -f agg-flogo.json
Creating initial project structure, this might take a few seconds ...
$ cd stream
$ flogo prepare
$ flogo build

stream/vendor/github.com/project-flogo/stream/pipeline

vendor/github.com/project-flogo/stream/pipeline/instance.go:21:13: undefined: channels.Channel
FATAL: command "build" failed: exit status 2

Please tell us about your environment (Operating system, docker version, browser & web ui version, etc):

Flogo version (CLI & contrib/lib. If unknown, leave empty or state unknown): 0.X.X
Performed update from GitHub today (26-Sep) of flogo-cli, flogo-contrib, and flogs-lib

Additional information you deem important (e.g. issue happens only occasionally):

flogo build error initContextImpl (missing Name method)

flogo build error
flogo/core version: v1.6.4

detail：
`

warn: [18:02:24.196] command exited with code 1: flogo build
error: [18:02:24.197] undefined
Error building project: # github.com/project-flogo/flow/definition
/root/gopath/pkg/mod/github.com/project-flogo/[email protected]/definition/definition_ser.go:237:32: cannot use ctx (type *initCtxImpl) as type activity.InitContext in argument to f:
*initCtxImpl does not implement activity.InitContext (missing Name method)
github.com/project-flogo/stream/pipeline
/root/gopath/pkg/mod/github.com/project-flogo/[email protected]/pipeline/stage.go:80:90: config.Operation undefined (type *StageConfig has no field or method Operation)
/root/gopath/pkg/mod/github.com/project-flogo/[email protected]/pipeline/stage.go:81:15: cannot use initCtx (type *initContextImpl) as type activity.InitContext in argument to f:
*initContextImpl does not implement activity.InitContext (missing Name method)

`

Sample for stream joining

Add a sample for joining multiple streams into a pipeline.

What is the Expected behaviour of sum function in a timeTumbling window type.

I am running the steams example from > Labs > Streams: Aggregate and I am curious about the expected behaviour when using a SUM style aggregate approach.

When I run the application from the example code it is working all fine and I can see the correct responses that at every 5000 ms it produces the correct response. However if I leave the app running and don't inject any messages I get no message at all. It seems the absence of invoking the http request does not create a result of zero. Can you confirm this is expected?

If it is expected what would be the recommend means for me to send to a downstream system that the result of 0 messages is 0 messages received. I am asking because in my use case I want to create a stream of data that I am using to track items and display the amount of items currently being tracked. However there will be times when there are 0 messages received in the time window. I need a way to notify downstream that I am still tracking but currently there are no items being tracked.

For example:
Time window 1: 1000 messages from 5 unique items = Result 5 Items being tracked
Time window 2: 1500 messages from 5 unique items = Still result 5 Items being tracked
Time window 3: 0 messages received so no unique items --> Currently no message. Therefore am I still tracking 5 or 0 items.

I need a way for a sum result to be available even if I receive no messages.

The version of the demo that I was using that now drives my question is detailed below:

"name": "aggregator",
"type": "flogo:app",
"version": "0.0.1",
"appModel": "1.0.0",

create myApp failed

Hello, when I run the command "flogo create -f agg-flogo.json myApp" on Ubuntu, it shows:

Creating initial project structure, this might take a few seconds ...
all dirs contained build errors
FATAL: command "create" failed: exit status 1