The collect node

%% input data ->

|json_emitter()
.every(500ms)
.json(
<<<{"code" : {"id": 224, "name" : "224"}, "message": "this is a test", "mode": 1}>>>,
<<<{"code" : {"id": 334, "name" : "334"}, "message": "this is another test", "mode": 1}>>>,
<<<{"code" : {"id": 114, "name" : "114"}, "message": "this is another test", "mode": 2}>>>,
<<<{"code" : {"id": 443, "name" : "443"}, "message": "this is another test", "mode": 1}>>>, 
<<<{"code" : {"id": 111, "name" : "111"}, "message": "this is another test", "mode": 1}>>>,
<<<{"code" : {"id": 443, "name" : "443"}, "message": "this is another test", "mode": 0}>>>,
<<<{"code" : {"id": 224, "name" : "224"}, "message": "this is another test", "mode": 0}>>>,
<<<{"code" : {"id": 111, "name" : "111"}, "message": "this is another test", "mode": 0}>>>,
<<<{"code" : {"id": 334, "name" : "334"}, "message": "this is another test", "mode": 0}>>>,
<<<{"code" : {"id": 551, "name" : "551"}, "message": "this is another test", "mode": 2}>>>,
<<<{"code" : {"id": 551, "name" : "551"}, "message": "this is another test", "mode": 0}>>>
)
.as('data')

%% collect 2 fields ('data.code' and 'data.message') with the key-field 'data.code'.
%% this node will output a data_batch item with a list of data-points
%% where the original timestamp and meta-data is preserved and
%% containing the fields mentioned before every 10 secondes

|collect()
%% the collect node will build an internal buffer with the value of the 'key_field' as index
.key_field('data.code')
%% criterion for adding a data-point to the internal collection buffer
.add(lambda: "data.mode" > 0)
%% criterion for removal of values
.remove(lambda: "data.mode" == 0)
%% we keep these fields in the resulting data_batch
.keep('data.code', 'data.message')
%% collection of type set, so no duplicates
.type('set')
.emit_every(10s)

%% make sense of the data-collection in counting the 'data.code' values
|aggregate()
.fields('data.code')
.functions('count')
.as('code_count')

|debug()