STREAM COMMANDS / QUERIES USING CLI
Stream commands – Stream is a feature in BangDB using which user can ingest, parse, process and analyse data in continuous manner.
CLI provides lots of commands to deal with streams in simple manner.
There are following important steps to set up streams.
1. Create Schema (using cli or can author json schema file separately)
2. Register the schema
3. Ingest and process data
1. Create Schema (using cli or can author json schema file separately)
2. Register the schema
3. Ingest and process data
Schema is a jsondocument which defines how the streams will get processed, stored and further the necessary actions may be taken.
The structure of json is very simple it is like following
Schema :
array of streams
array of notification templates
array of user defined functions [ UDFs ]
Stream :array of notification templates
array of user defined functions [ UDFs ]
array of attributes
array of computed attributes [ attrs that could be computed at run time and added to the stream ]
array of referred attributes [ attrs that could be referred and added to the stream ]
array groupby that could be computed and updated continuously
array of joins that could result in data streamed into another derived streams
array of filter that could send filtered data to another derived streams
array of CEP queries that could run continuously to find patterns, output could be sent to another derived streams
array of entity long term stats to be computed continuously
array of computed attributes [ attrs that could be computed at run time and added to the stream ]
array of referred attributes [ attrs that could be referred and added to the stream ]
array groupby that could be computed and updated continuously
array of joins that could result in data streamed into another derived streams
array of filter that could send filtered data to another derived streams
array of CEP queries that could run continuously to find patterns, output could be sent to another derived streams
array of entity long term stats to be computed continuously
See Stream Processing for more details
Let’s see a sample schema
{
"schema":"myschema",
"streams":[
{
"name":"stream1",
"type":1,
"swsz":86400,
"inpt":[
],
"attr":[
{
"name":"a",
"type":5,
"kysz":32,
"sidx":1,
"ridx":1,
"stat":1
},
{
"name":"b",
"type":5,
"kysz":32,
"stat":2
},
{
"name":"g",
"type":11,
"stat":3
},
{
"name":"h",
"type":11
}
],
"catr":[
{
"name":"m",
"type":9,
"opnm":"ADD",
"stat":3,
"iatr":[
"b",
"a"
]
},
{
"name":"mexp",
"type":9,
"opnm":"MATH_EXP",
"iatr":[
"((($g+$h)*2)+($g*$h))"
]
}
],
"gpby":[
{
"gpat":[
"a",
"b"
],
"iatr":"g",
"stat":3,
"gran":3600,
"kysz":48
}
],
"refr":[
{
"name":"myrefr1",
"iatr":[
"c"
],
"rstm":"stream2",
"ratr":[
"p1"
],
"jqry":{
"cond":[
"b"
],
"opnm":"ATTR",
"args":[
"b"
]
}
}
]
},
{
"name":"stream2",
"type":1,
"inpt":[
],
"attr":[
{
"name":"p1",
"type":5,
"kysz":24,
"sidx":1
},
{
"name":"b",
"type":5,
"kysz":32
}
]
}
]
}Other example could be following;This runs CEP in continuous manner to find a pattern where for two stocks, the current value is exactly same but the stocks are different
{
"schema":"test",
"streams":[
{
"name":"s1",
"type":1,
"swsz":86400,
"inpt":[
],
"attr":[
{
"name":"stock",
"type":5,
"kysz":16
},
{
"name":"price",
"type":9
}
],
"cepq":[
{
"name":"mystock",
"type":1,
"tloc":3000,
"ratr":[
"price"
],
"rstm":"s1",
"iatr":[
"stock",
"price"
],
"jqry":{
"cond":[
"stock",
"price"
],
"opid":11,
"args":[
"stock",
"price"
],
"cmp":[
"NE",
"EQ"
],
"seq":1
},
"fqry":{
"name":"{\"query\":[{\"key\":\"price\", \"cmp_op\":0, \"val\":70.5}]}",
"type":1
},
"cond":[
{
"name":"NUMT",
"val":1,
"opid":1
},
{
"name":"DUR",
"val":1000,
"opid":0
}
],
"ostm":"s2"
}
]
},
{
"name":"s2",
"type":3,
"swsz":86400,
"inpt":[
"s1"
],
"attr":[
{
"name":"stock",
"type":5,
"kysz":16
},
{
"name":"price",
"type":9
}
]
}
]
}And other example, which joins two streams in continuous manner.This joins temperature and pressure for any given point and sends the joined data to temp_pressure_stream
{
"schema":"myschema",
"streams":[
{
"name":"temp_stream",
"type":1,
"swsz":81600,
"inpt":[
],
"attr":[
{
"name":"temp",
"type":11
},
{
"name":"point",
"type":9
}
],
"join":[
{
"name":"temp_pressure_join",
"type":3,
"tloc":3000,
"iatr":[
"temp",
"point"
],
"rstm":"pressure_stream",
"ratr":[
"pressure"
],
"jqry":{
"cond":[
"point"
],
"opid":11,
"args":[
"point"
]
},
"ostm":"temp_pressure_stream"
}
]
},
{
"name":"pressure_stream",
"type":1,
"inpt":[
],
"attr":[
{
"name":"pressure",
"type":11
},
{
"name":"point",
"type":9
}
],
"join":[
{
"name":"temp_pressure_join",
"type":5,
"tloc":3000,
"ratr":[
"temp",
"point"
],
"rstm":"temp_stream",
"iatr":[
"pressure"
],
"jqry":{
"cond":[
"point"
],
"opid":11,
"args":[
"point"
]
},
"ostm":"temp_pressure_stream"
}
]
},
{
"name":"temp_pressure_stream",
"type":3,
"inpt":[
"temp_stream",
"pressure_stream"
],
"attr":[
{
"name":"point",
"type":9
},
{
"name":"temp",
"type":11
},
{
"name":"pressure",
"type":11
}
]
}
]
}Stream Commands – To register the given schema
We can write a schema (like above) and directly register it with the server using cli
Suppose the schema is in file “sample_schema.json” then we can register as below
register schema sample_schema.json @@@@ time taken to init stream [ myschema__stream2 ] is ... Usage : ------------------------- | Time : 0.003 msec | ------------------------- User : 0.01 msec Sys : 0 msec ------------------------- @@@@ time taken to init stream [ myschema__stream1 ] is ... Usage : ------------------------- | Time : 89.097 msec | ------------------------- User : 89.975 msec Sys : 1 msec ------------------------- success
Stream Commands – To de-register the schema, drop the schema
deregister schema myschema
success
OR, we can create using the workflow provided by the cli, if we issue command “create schema …”
create schema myschema do you wish to read earlier saved schema for editing/adding? [ yes | no ]: no
Select “no” as we are creating from scratch. We should select yes if we have some schema on the disk and we would like to add/edit to it
Then it asks to select the stream nam/me
what's the name of the stream that you wish to add?: stream1
We input the name of the stream “stream1” and the press enter
Post this we must select the type of the stream
what's the type of the stream [ raw/normal primary (1) | fltr output (2) | join/cep output (3) | Entity (4) | skip (5) ] (or Enter for default (1)):1
There are few types of streams, raw is the one where we ingest data and all others are derived streams, i.e. used by the stream processing engine flter output stream is one where filtered data goes, join/cep is the one where joined data goes and entity stream is the one where long term aggregate goes.
There are few other types, such as stat, groupby, buffer etc. but they are all abstracted and used by the stream processing engine solely
There are few other types, such as stat, groupby, buffer etc. but they are all abstracted and used by the stream processing engine solely
Since this is the primary/raw stream hence we will select 1
Now, define the size of the sliding window, enter for default one day
what’s the size of the sliding window in seconds [ or enter for default 86400 sec (a day) ]:
By now, we have empty stream schema defined, now we must define what goes inside the schema framework
enter 1 to add attributes first
What would you like to add (press Enter when done) [ attr (1) | catr (2) | refr (3) | gpby (4) | fltr (5) | join (6) | entity (7) | cep (8) | notifs (9) ]:1
This is start the sub workflow to add attributes
add attributes...attribute name: a attribute type (press Enter for default(5)) [ string(5) | long(9) | double (11) ]: 5 enable sidx [ 0 | 1 ]: 1 attribute key size in bytes (press Enter for default(24 bytes)): 32 enable ridx [ 0 | 1 ]: 1 enable stat [ none(0) | count (1) | unique count (2) ]: 1add another attribute ? [ yes | no ]:
to add any attribute,
type: we can select type of attributes
sidx: to create secondary index on the attribute. Good for query and also if this is going to be used in join or cep or filter. Then we should enable the sidx
key size : only when attribute type is string then we need to select this
ridx: reverse index, switch it on if we wish to do search on this
stat: this is generic running stats. For string type we can do count or unique count. For double or long we can also select stat as 3,
which is complete set of stats [ count, min, max, avg, skey, stddev, excurt etc… ] We can keep following the sub workflow until we have created enough attributes
type: we can select type of attributes
sidx: to create secondary index on the attribute. Good for query and also if this is going to be used in join or cep or filter. Then we should enable the sidx
key size : only when attribute type is string then we need to select this
ridx: reverse index, switch it on if we wish to do search on this
stat: this is generic running stats. For string type we can do count or unique count. For double or long we can also select stat as 3,
which is complete set of stats [ count, min, max, avg, skey, stddev, excurt etc… ] We can keep following the sub workflow until we have created enough attributes
Now let’s also create another stream “stream2” and add two attributes p1 and b as shown above, following the same workflow
let’s now create “catr”
When creating “catr”, it starts sub workflow for the “catr” itself
add computed attributes (catr)...attribute name (press Enter to end): m attribute type (press Enter for default (5)) [ string(5) | long(9) | double (11) ]: 9 available default ops are [ COPY | ADD | MUL | DIV | PERCENT | SUB | UPPER | LOWER | COPY_VAL | LOG_E | LOG_2 | LOG_10 | MATH_EXP | PRED | TS | YEAR | YEAR_EPOCH | MONTH | MONTH_EPOCH | WEEK | WEEK_MONTH | WEEK_EPOCH | DAY | DAY_WEEK | DAY_MONTH | DAY_EPOCH | HOUR | HOUR_EPOCH | MINUTE | MINUTE_EPOCH | SECOND | ABS ] or Enter custom udf nameenter the name of the intended operation from the above default ops (press Enter to end): ADD
The catr is about computing new attribute based on some logic/computations/operations.
Here we have 20+ default operations available, if we want to do something beyond these then we can write our own UDF and use the same.
Here we have 20+ default operations available, if we want to do something beyond these then we can write our own UDF and use the same.
Select “ADD” as the ops here, so we wish to add something.
We can ADD attributes or some fixed val.
Here let’s say we wish to add the attributes, therefore we need to select attributes that we wish to add
enter the name of the intended operation from the above default ops (press Enter to end): ADD enter the input attributes on which this ops will be performed, (press Enter once done): b enter the input attributes on which this ops will be performed, (press Enter once done): a enter the input attributes on which this ops will be performed, (press Enter once done): enter sequence [ 0 | 1 ], if 1 then it will be done before refer else post refr:1