When using the Data Lake sink, the incoming events are stored in an InfluxDB.
Implementation
org.apache.streampipes.sinks.internal.jvm.datalake
The concrete implementation comprises a Data Lake class, a Data Lake Controller class, a Data Lake InfluxDB Client class and a Data Lake Parameters class. The code is basically the same as for the InfluxDB sink (org.apache.streampipes.sinks.databases.jvm.influxdb).
Data Lake Parameters Class
The parameter class defines the necessary parameters for the configuration of the sink.
parameter | description |
---|---|
influxDbHost | hostname/URL of the InfluxDB instance. (including http(s)://) |
influxDbPort | port of the InfluxDB instance |
databaseName | name of the database where events will be stored |
measureName | name of the Measurement where events will be stored (will be created if it does not exist) |
user | username for the InfluxDB server |
password | password for the InfluxDB server |
timestampField | field which contains the required timestamp (field type = http://schema.org/DateTime) |
batchSize | indicates how many events are written into a buffer, before they are written to the database |
flushDuration | maximum waiting time for the buffer to fill the Buffer size before it will be written to the database in ms |
dimensionProperties | list containing the tag fields (scope = dimension property) |
Data Lake Controller Class
In controller class, the model is declared for viewing and configuration in Pipeline Editor, and initializes sink on invocation of pipeline.
The measurement name and the timestamp field are derived from user input, the remaining parameters (except batch size and flush duration) from org.apache.streampipes.sinks.internal.jvm.config.SinksInternalJvmConfig. Batch size is fixed to 2000 events and flush duration is set to 500 ms.
Data Lake Class
The data lake class itself essentially controls the saving of events to the database. For this purpose, it uses the Data Lake InfluxDB Client.
method name | description |
---|---|
onInvocation | starting the DataLakeInfluxDbClient, registering and initializing new measurement series in InfluxDB |
onEvent | adding empty label field to incoming event and storing event in database |
onDetach | stopping the DataLakeInfluxDbClient |
Image data, unlike events, is not stored directly in database but as Image files in a corresponding directory (writeToImageFile).
In addition, the class contains two utility methods (registerAtDataLake and prepareString)
Data Lake InfluxDB Client Class
Client class that connects to InfluxDB and writes events directly to database. Uses the Data Lake Parameters described above.
method name | description |
---|---|
validate | checks whether the influxDbHost is valid |
connect | connects to the InfluxDB server, sets the database and initializes the batch-behaviour |
databaseExists | checks whether the given database exists |
createDatabase | creates a new database with the given name |
save | saves an event to the connnected InfluxDB database |
stop | shuts down the connection to the InfluxDB server |
TODO:
- validate(): use validation method (org.apache.commons.validator.routines.InetAddressValidator) instead of regex check
REST API
DataLakeNoUserResourceV3
org.apache.streampipes.rest.impl.datalake
This class contains the basic interface definition for setting up a new measurement series in Data Lake and calls the underlying methods of org.apache.streampipes.dataexplorer.DataLakeNoUserManagementV3. Usage of the related API calls does not require any authentification with valid username and password.
method name | request type | path | description |
---|---|---|---|
addDataLake | POST | /{measure} | adds new measurement series with specified measure name and related event properties (column names) in InfluxDB |
TODO:
- [STREAMPIPES-348]: fix issue with special characters in user-defined measure name
- add authentication obligation to addDataLake method
DataLakeResourceV3
org.apache.streampipes.ps
This class contains the extended interface definition and calls the underlying methods of org.apache.streampipes.dataexplorer.DataLakeManagementV3 and org.apache.streampipes.dataexplorer.utils.DataExplorerUtils when invoked. Usage of below mentioned API calls requires authentification with valid username and password.
method name | request type | path | description |
---|---|---|---|
getPage | GET | /data/{index}/paging | returns pages with predefined number of events per page of a specific measurement series from InfluxDB |
getAllInfos | GET | /info | returns list with ids of all existing measurement series (including event schema) from InfluxDB |
getAllData | GET | /data/{index} /data/{index}/last/{value}/{unit} /data/{index}/{startdate}/{enddate} | returns all stored events of a specific mesurement series from InfluxDB |
getAllDataGrouping | GET | /data/{index}/{startdate}/{enddate}/grouping/{groupingTag} | returns all events within a specified time frame of a specific mesurement series grouped by a specific tag from InfluxDB |
removeAllData | DELETE | /data/delete/all | removes all stored events from InfluxDB |
downloadData | GET | /data/{index}/download | downloads all events of a specific mesurement series from InfluxDB in desired format |
getImage | GET | /data/image/{route}/file | returns png image from file route |
saveImageCoco | POST | /data/image/{route}/coco | stores image as file at file route |
getImageCoco | GET | /data/image/{route}/coco | returns image at file route as application/json |
labelData | POST | /data/{index}/{startdate}/{enddate}/labeling/{column}/{timestampColumn}?label= | updates label in specified column for all events within specified time frame to provided label value |
TODO:
- fix export of data from data lake, which currently returns two timestamp fields
- extend aggregation functionality to support non-numeric values (e.g. strings → majority vote) add the possibility to specify an aggregation function
- in general: alignment of the single endpoint definitions and consideration of the extensions below
Ideas for possible adaptations and extensions of the REST API
In addition to the TODOs listed above in the text, the following adjustments and enhancements might be worth considering. Thereby, it is important that the implementation of the endpoints is as independent as possible from the technology of the data lake (e.g. avoiding InfluxDB-specific methods).
- Extension of the remove endpoint by the capability to
- selectively delete an individual measurement series
- delete measurements of a measurement series within a specific time interval or before a specific date
- Adding an edit endpoint for adjusting data lake specific properties such as retention time.
Both extensions could be included in a kind of data management tool in the UI within an admin view (e.g. in the manner of the pipeline view).
Another possible adaptation would be the comprehensive implementation of an append-only approach for time series data. In particular, the functionality of the labelData method would have to be adapted here, which currently works with updates of existing DB entries.
References:
- Apache StreamPipes Documentation: https://streampipes.apache.org/docs/docs/pe/org.apache.streampipes.sinks.internal.jvm.datalake/
9 Comments
Johannes Tex
Hi,
here are some comments from my side.
Data Lake Controller Class
DataLakeResourceV3
the endpoints have no authentication because they are called from the DataLakeSink and of course the user/password is not known there. In order to introduce authentication, a concept with technical users would probably have to be introduced.
Data Lake Controller Class
Here we use a lot of path parameters, which are query parameters. E.g.
/data/{index}/{startdate}/{enddate}/grouping/{groupingTag} → /data/{index}?stardate=...&enddate=..&groupby=...
General - Saving and Reading Images and Coco Files
Daniel Ebi
Hi Johannes,
thanks for your thoughts on possible adaptions / extentions.
Especially the configurability of flush duration and batch size is a good idea that we defintely should include.
Also your point with the introduction of a technical user is a good point. Maybe in this context it also makes sense to think about extending the token-based authentication (service API key), which is already used for "external" services. This mechanism could be used to implement token-based authentication at the Data Lake sink instead of using a technical user. The token would then have to be set as an environment variable, for example. What do you think about that solution? Or would it be better to use an explicit technical user?
Dominik Riemer
Hi Daniel,
thanks for coming up with this!
Regarding authentication, I'm currently working on improving the communication between pipeline elements and core services. The idea is to provide an instance of the StreamPipes client within the onInvocation method so that pipeline elements requesting anything from the core APIs can directly interact with these using the aforementioned service token. In this case, the data lake sink would just call the platform service API in order to create a new measure or do anything else in the core (e.g., sending emails or creating live visualizations). For that, we need to extend our Auth concept with a service user. I'm also trying to get rid of the Consul config provider and integrate this directly into the description of a pipeline element.
After that, we can simply move the data lake API to the platform services module and make the endpoint authenticated.
In the meanwhile, we can work on improving the REST interface itself, e.g., the comment from Johannes concerning the usage of query parameters instead of path parameters seems very important to me. So discussing how to improve the DataLakeRestResourceV3 (let's say the way towards v4) should be the next step IMO.
From a user perspective, it should be easily understandable how to get data, filter and group data, upload/download/list images and so on. Do you have any first suggestion on how to redesign this endpoint? Otherwise, I can provide some first input.
Dominik
Daniel Ebi
Hi,
I have created a first draft for the revised endpoint definitions (see table below). Thereby, I have tried to map as suggested as much as possible via query parameters. However, the user should have for example the possibility to specify multiple grouping tags for the "groupBy" parameter.
The file sharing part for the image / coco files I kept very prototypical and assumed a simple object store (key-value store). Here we can also discuss the solution mentioned by Johannes using a blob storage and adapt the draft accordingly.
I am interested in your comments and suggestions for adjustments.
Suggestion for the revised endpoint definitions
-
start date, if not specified: first element
end date, if not specified: last element
grouping tags
aggregation function (e.g. mean)
time interval for aggregation (e.g. 1m - one minute), if specified: performing group by time
getting specific page with predefined number of events per page of a specific measurement series
page number
data points per page
downloading events of a specific mesurement series from in desired format
download format (csv, json)
start date, if not specified: first element
end date, if not specified: last element
start date, if not specified: first element
end date, if not specified: last element
index of measurement series
start date, if not specified: first element
end date, if not specified: last element
label column
class label
file type (image, COCO)
Johannes Tex
Hi Daniel,
that looks already great
I would suggest changing the API paths a little to make it a little more RESTful:
request type
path
query parameters
-
What you think about a "limit" query param?
What is the difference to /measurement/{index}/data ?
Maybe it is possible to use the same API, just add the format query param?
Dominik Riemer
Looks good!
A few comments/questions from my side:
I'm not sure whether the multiple query endpoints for paged, grouped and non-grouped queries return the same schema - that would be something we would need to harmonize in case there is a single endpoint.
Regarding the comment from Johannes, what is the difference between measurement and index?
And regarding naming of resources as raised by Johannes, a common best practice is to have an endpoint such as /measurements (using plural) where a GET would return a list of all measurements and /measurements/{measurementId} (maybe this would be index) that returns the content for the specific source. In that case, to delete a specific measurement, a DELETE call to /measurements/{measurementId} could be available. Hope this helps!
And another, more high-level comment regarding the roadmap: In the near future, I'd really like to simplify some StreamPipes concepts. One idea here is that everything in StreamPipes is either a Data Set or a Data Stream, which can be configured to be persisted (without manually creating a pipeline). For pipelines that transform data streams that should be stored in the data lake, there would be two "Virtual Data Stream/Set" sinks instead of the data lake/dashboard sink, which itself represent a stream/set in the UI that can be configured to be persisted. In this case, any resource in the data lake could be discovered by the (e.g., appId or some other identifier) of the underlying data stream/set. Maybe we can already foresee that in the naming of resources for this API redesign.
Daniel Ebi
Thanks for your helpful input. I have incorporated your comments into the draft (see table below).
As suggested, I have left the file sharing /storage part out for now. However, this is also an exciting and important topic that we should address.
@Dominik: As far as I know, the image upload is currently done via the image connect adapter (set: org.apache.streampipes.connect.adapters.image.set, stream: org.apache.streampipes.connect.adapters.image.stream). I have not worked with it yet. But it would certainly be nice to have an endpoint for it.
Regarding the high-level roadmap you mentioned, we can foresee the use of an UUID (or similar) as measurementID. However, imo it would then make sense to write additional metadata about the measurement series to the data lake when creating a persistent adapter (e.g. user-assigned plaintext name), for user convenience purposes. But I think, with InfluxDB only measurement-level metadata is supported in the form of tags, which would generate too much overhead if this additional information had to be written each time. But please correct me here if I am wrong. But does the resource naming fit then or do you have a concrete suggestion for the renaming?
-
-
slicing data by timestamp criterion
start date, if not specified: first element
end date, if not specified: last element
paging
page number
limit rows per page
specify offset (time_offset)
grouping
grouping tags (comma-separated)
aggregation function (e.g. mean)
time interval for aggregation (e.g. 1m - one minute), if specified: performing group by time
downloading
can be combined with slicing operators
data format (csv, json)
start date, if not specified: first element
end date, if not specified: last element
POST
start date, if not specified: first element
end date, if not specified: last element
label column
class label
Dominik Riemer
Hi Daniel,
thanks, looks very good to me!
I'd say we can keep the user-defined index name as the measurementId by now and switch it to the unique ID of the corresponding data stream once we've integrated data streams and their persistence. In that case, the metadata would directly come from the data stream description and users would only interact with the name of the data stream in the data explorer and dashboard without the need to explicitly assign/know the index name.
So from my point of view, this API redesign is ready for entering the implementation phase
Johannes Tex
From my side as well