Data analysis
Data store & search concepts and components
After your data sended to Live Objects using data streams, Each source is processed by one or many components. Data storage & search is then relies upon:
-
the store service aims to store data messages from devices (devices, gateways, IoT app collecting data, etc.) as time-series data streams
-
the indexing service aims to index data messages collected from devices (devices, gateways, IoT app collecting data, etc.) by using model field after injected by the previous components.
-
the search service based on the popular open-source Elasticsearch product.
Store service
The REST interface allows to add data to a stream and to retrieve data from a stream. A stream could be for example associated to a unique of device (streamID could be therefore a device Identifier) or one type of data coming from a device (streamID could be therefore in this format deviceIdentifier-typeOfData)
Manage your data streams
Every source publishing or injecting the data into Live Objects (Device, gateway or business application), must set the stream which would be where you want to store your data.
In order to be able to use and process the data collected correctly, it is necessary to be able to read it and then store it. The management of the read and write processes in Live Objects is done through data stream channels.
A data stream is associated with each publisher and it can be a device with a single interface or several, you must take into consideration where the published messages will be written.
By default, each provisioned device, Live Objects associates a streamId with the defaulDataStreamId whose value is set automatically by Live Objects. The messages will be written under this stream.
So, to correctly manage the streams of your device, you should either use the stream provided by default or modify it if necessary.
When you want the device to use a particular streamId, the publisher must indicate in each uplink the desired stream. This allows data to be writed and stored as needed.
Default data stream payload :
{
"value": {"temp":23.8},
"model": "data_model_v1"
}Your data are stored under the defaultDataStreamId value stream. In this case the value is a device URN (or the value that you set in defaultDataStreamId field).
Custom data stream payload :
{
"streamId" : "my_stream_1"
"value": {"temp":24.1},
"model": "data_model_v1"
}To change your Custom data stream :
{
"streamId" : "my_stream_2"
"value": {"temp":20.1},
"model": "data_model_v1"
}| In the device side, make sure to set the device’s custom streamId properly (set correctely the payload published by the device), as it determines which stream the message will be published to. Otherwise, messages will be sent to the configured/defaultStreamId (configure it in Live objects UI) |
Add a data message to a stream
Request:
POST /api/v0/data/streams/{streamId}
X-API-Key: <your API key>
Accept: application/json
body param |
description |
data |
JSON object conforming to data message structure |
Warning: the streamId is provided as the last segment of the url.
Example:
POST /api/v0/data/streams/myDeviceTemperature
{
"value": {"temp":24.1},
"model": "data_model_v0"
}For this example, the "value.temp" field of model "data_model_v0" will be defined as a double type. If a String type is used in the future for "value.temp", a new model must be defined. In case that "value.temp" is set a String type with model "data_model_v0", the message will be dropped by the search service.
Add an encoded data message to a stream
In order to use the decoding capability of Live Objects, a DataMessage must contains additional 'value.payload' and 'metadata.encoding' fields :
Example:
POST /api/v0/data/streams/myDeviceTemperatureAndPressure
{
"value": {
"payload": "000003F5000000DD"
},
"metadata": {
"encoding": "twointegers"
}
}| Field | Description |
|---|---|
value.payload |
(Mandatory). Payload to decode. In case of binary content, HexBinary String representation of the payload to decode. |
metadata.encoding |
(Mandatory). Encoded format name, that should match the 'encoding' name of the decoder that can process this message. |
All other fields of DataMessage (timestamp, model, location, tags…) can also be set in the encoded DataMessage.
You can also set the encoding property in the device’s interface definition in order to force metadata.encoding value for all data messages sent from this device through its interface.
Add a bulk of data messages
Request:
POST "/api/v0/data/bulk X-API-Key: <your API key> Accept: application/json
body param |
description |
data |
JSON array conforming to an array of data message structure |
A bulk will be processed if all arrays elements are valid, otherwise the bulk will be rejected. Maximum size of the bulk is 1000.
Warning : the streamId is mandatory for each element of the Bulk. This is a difference with the REST API for adding data to a stream.
Example:
POST /api/v0/data/bulk
[
{
"streamId" : "temperature_stream_1"
"value": {"temp":24.1},
"model": "data_model_v1"
},
{
"streamId" : "temperature_stream_1"
"value": {"temp":24.1},
"model": "data_model_v1"
},
{
"streamId" : "pressure_stream_1"
"value": {"pressure":1024.0},
"model": "data_model_v1"
}
]Retrieve data from a stream
Request:
GET /api/v0/data/streams/{streamId}
X-API-Key: <your API key>
Accept: application/json
Query params |
Description |
limit |
Optional. max number of data to return, value is limited to 100 |
timeRange |
Optional. filter data where timestamp is in timeRange "from,to" |
bookmarkId |
Optional. id of document. This id will be used as an offset to access to the data. |
Documents are provided in reverse chronological order (newest to oldest).
Example:
GET /api/v0/data/streams/myDeviceTemperature
{
"id": "57307f6c0cf294ec63848873",
"streamId": "myDeviceTemperature",
"timestamp": "2016-05-09T12:15:41.620Z",
"model": "temperature_v0",
"value": {
"temp": 24.1
},
"created": "2016-05-09T12:15:40.286Z"
}Indexing service
After the data collected it may be associated to a model. The model is a fundamental concept for the search service, it specifies the schema of the JSON "value" object. The model is dynamically updated based on the data injected.
The model concept is necessary to avoid mapping conflicts in the underlying elasticsearch system.
Model field in data message
The model allows to give pattern for the value object, A model can be seen as a "mapping space" in which data types must not conflict. Some precautions must be taken when using model field :
-
If the model is not provided, "value" object will be not indexed by the search service. Nevertheless, the data will be stored in the store service and all information except value object will be indexed in search service.
-
If the value JSON object does not comply with the provided model (for example, a field change from Integer type to String type), the data will be not inserted in the search service. The data message will be only stored in the store service.
| Without a model, you will not be able to search or aggregate on any value.* field. |
Set model field to index your data
To make the messages sent by devices indexed in data store you must setting your data model with the following steps :
-
when you provision your decoder set "model" field in your data model.
-
when the "model" field is set and elasticsearch (ES) is enable, you can make elasticsearch queries on value fields only after the messages are decoded by your decoder and indexing service done.
-
each field of value collection is indexed by search service. In order to be able to search the message, copy this 'value' in a @model.value field which is analyzed by Elasticsearch. The message is consistent with others messages having the same model.
| Max total field number which can be indexed in ES : the limit allowed per tenant is max 1000 fields. |
exemple : data message sent by device and shown the indexed fields with values.
{
"timestamp" : "2018-05-31T16:15:21.288Z,"
"streamId" : "urn:lo:nsid:mqtt:{deviceId},"
"model": "my_encoding_model_v0"
"value": {
"temp": 12.75,
"humidity": 62.1,
"gpsFix": true,
"gpsSats": [12, 14, 21]
}
}Indexed field :
Field |
type |
temp |
float |
humiditity |
float |
gpsFix |
boolean |
gpsSats |
integer |
Search service
The REST request body search API is provided to perform search queries.
To learn more about the search API, read the Exploring your Data section of Elasticsearch: The Definitive Guide.
To perform a search query
Full search
Request
POST /api/v1/data/search X-API-Key: <your API key> Accept: application/json
parameter |
description |
dsl request (body) |
elasticsearch DSL request |
Elasticsearch 7
A major upgrade of our search engine from Elasticsearch 5 to Elasticsearch 7 has been made in 2021. The search request DSL body parameter will follow the Elasticsearch 7 one.
There are a few breaking changes compared to the previous one based on version 5, mainly some deprecated operators or changes in naming.
| You need to ensure you are not currently using these specific operators before the change will apply. |
You will receive a HTTP 400 response in such a case.
Below are the main changes :
-
Queries on boolean fields now strictly parse boolean-like values. This means only the strings "true" and "false" will be parsed into their boolean counterparts.
-
The 'in' query (a synonym for the 'terms' query) has been removed
-
Support for empty query objects ({ }) has been removed from the query DSL
-
The deprecated 'geo_distance_range' query has been removed
-
The 'all_fields' parameter for the 'query_string' and 'simple_query_string' has been removed. Set 'default_field' to '*' instead.
-
The deprecated 'prefix' parameter (a synonym for the 'value' parameter) of the 'prefix' query has been removed.
-
The deprecated 'le' (a synonym for 'lte') and 'ge' (a synonym for 'gte') parameter of the range query have been removed.
The full list can be found here (refer to section 'Search and Query DSL changes') :
The response of /api/v0/data/search request will stay unchanged.
But the new API version /api/v1/data/search uses the new default Elasticsearch 7 behavior : hits.total is limited to 10.000 documents; and the field is now a JsonObject instead of a number :
{
"hits": {
"total": 10123,
...
}
}{
"hits": {
"total": {
"value": 1000,
"relation": "eq"
},
...
}
}
{
"hits": {
"total": {
"value": 10000,
"relation": "gte"
},
...
}
}In this new v1 API, you can also use the query parameter trackTotalHits=true to force a full count (i.e. that can exceed 10.000).
Example
This query requests statistics from the myDeviceTemperature stream temp field.
Request:
POST /api/v1/data/search
{
"size": 0,
"query": {
"term": {
"streamId": "myDeviceTemperature"
}
},
"aggs":
{
"stats_temperature": {
"stats": {
"field": "@temperature_v0.value.temp"
}
}
}
}If a model has been provided, search query must be prefixed by @<model>: @temperature_v0.value.datapath
Response:
{
"took": 1,
"hits": {
"total": {
"value": 2,
"relation": "eq"
}
},
"aggregations": {
"stats_temperature": {
"count": 2,
"min": 24.1,
"max": 25.9,
"avg": 25,
"sum": 50
}
}
}Hits search
To perform the same search query; but with the 'hits' part extracted and JSON formated as an array of data messages (to use when you are only interested in the 'hits' part of Elasticsearch answer):
Request
POST /api/v1/data/search/hits X-API-Key: <your API key> Accept: application/json
parameter |
description |
dsl request (body) |
elasticsearch DSL request |
Example
This query requests last data for all devices using the model: temperature_v0.
Request:
POST /api/v0/data/search/hits
{
"size": 10,
"query": {
"term": {
"model": "temperature_v0"
}
}
}Response:
[
{
"id": "57308b3b7d84805820b35345",
"streamId": "myDeviceTemperature",
"timestamp": "2016-05-09T13:06:03.903Z",
"model": "temperature_v0",
"value": {
"temp": 25.9
},
"created": "2016-05-09T13:06:03.907Z"
},
{
"id": "573087777d84805820b35344",
"streamId": "myDeviceTemperature",
"timestamp": "2016-05-09T12:49:59.966Z",
"model": "temperature_v0",
"value": {
"temp": 24.1
},
"created": "2016-05-09T12:49:59.977Z"
},
{
"id": "5730b1577d84805820b35347",
"streamId": "myStreamDemo-temperature",
"timestamp": "2016-05-09T15:48:39.390Z",
"model": "temperature_v0",
"value": {
"temp": 24.1
},
"created": "2016-05-09T15:48:39.395Z"
}
]Geo Queries
Geo Query can be performed through all fields with name matching location (case insensitive).
In order to geoquery these fields, you must add @geopoint to the location query path: location@geopoint.
Request:
POST /api/v1/data/search/hits
{
"query": {
"bool" : {
"must" : {
"match_all" : {}
},
"filter": {
"geo_distance": {
"distance": "10km",
"location.@geopoint": {
"lat": 43.848,
"lon": -3.417
}
}
}
}
}
}Response:
[
{
"id": "57308b3b7d84805820b35345",
"streamId": "myDeviceTemperature",
"location": {
"lat": 43.8,
"lon": -3.3
}
"timestamp": "2016-05-09T13:06:03.903Z",
"model": "temperature_v0",
"value": {
"temp": 25.9
},
"created": "2016-05-09T13:06:03.907Z"
}
]Search Query samples
Here are some query samples that can be used.
Aggregations are very useful to retrieve data grouped by any criteria: list all known tags, get all last value per stream, get mean temperature per tag, get the list of streams that have not send data since a date…
The aggregations results are stored as 'buckets' in the result.
You can also add filters (geoquery, wildcards, terms…) to all your aggregations query to target specific 'buckets' or data.
Give me all you got !
Request:
{
"query": {
"match_all": {}
}
}Give me the list of all known tags
Request:
{
"size": 0,
"aggs": {
"grouped_by_tags": {
"terms": {
"field": "tags"
}
}
}
}Response:
{
"took": 44,
"hits": {
"total": {
"value": 66,
"relation": "eq"
}
},
"aggregations": {
"grouped_by_tags": {
"doc_count_error_upper_bound": 0,
"sum_other_doc_count": 0,
"buckets": [
{
"key": "tag_1",
"doc_count": 53
},
{
"key": "tag_2",
"doc_count": 13
}
]
}
}
}Give me the last value of all my streams
Request:
{
"size":0,
"aggs": {
"tags": {
"terms": {
"field": "streamId"
},
"aggs": {
"last_value": {
"top_hits": {
"size": 1,
"sort": [
{
"timestamp": {
"order": "desc"
}
}
]
}
}
}
}
}
}Response:
{
"took": 19,
"hits": {
"total": {
"value": 11,
"relation": "eq"
}
},
"aggregations": {
"tags": {
"doc_count_error_upper_bound": 0,
"sum_other_doc_count": 0,
"buckets": [
{
"key": "device_1",
"doc_count": 7,
"last_value": {
"hits": {
"total": 7,
"max_score": null,
"hits": [
{
...
}
]
}
}
},
{
"key": "device_2",
"doc_count": 123,
"last_value": {
"hits": {
"total": 123,
"max_score": null,
"hits": [
{
...
}
]
}
},
...
}
]
}
}
}Give me the list of devices that have not send data since 2017/03/23 10:00:00
Request:
{
"size":0,
"aggs": {
"tags": {
"terms": {
"field": "streamId"
},
"aggs": {
"last_date": {
"max": {
"field": "timestamp"
}
},
"filter_no_info_since": {
"bucket_selector": {
"buckets_path": {
"lastdate":"last_date"
},
"script": {
"inline": "lastdate<1490263200000",
"lang":"expression"
}
}
}
}
}
}
}Response:
{
"took": 8,
"hits": {
"total": {
"value": 9,
"relation": "eq"
}
},
"aggregations": {
"tags": {
"doc_count_error_upper_bound": 0,
"sum_other_doc_count": 0,
"buckets": [
{
"key": "device_12",
"doc_count": 7,
"last_date": {
"value": 1489504105020,
"value_as_string": "2017-03-14T15:08:25.020Z"
}
},
{
"key": "device_153",
"doc_count": 2,
"last_date": {
"value": 1489049619254,
"value_as_string": "2017-03-09T08:53:39.254Z"
}
}
]
}
}
}