NGSIArcgisFeatureTableSink
Content:
Functionality
com.iot.telefonica.cygnus.sinks.NGSIArcgisFeatureTableSink is a sink designed to persist NGSI-like context data events within an [ArcGis] (https://www.arcgis.com/home/index.html) feature table. Usually, such a context data is notified by a Orion Context Broker instance, but could be any other system speaking NGSI language.
Independently of the data generator, NGSI context data is always transformed into internal NGSIEvent objects at Cygnus sources. In the end, the information within these events must be mapped into specific ArcGis structures.
Next sections will explain this in detail.
Mapping NGSI events to NGSIEvent objects
Notified NGSI events (containing context data) are transformed into NGSIEvent objects (for each context element a NGSIEvent is created; such an event is a mix of certain headers and a ContextElement object), independently of the NGSI data generator or the final backend where it is persisted.
This is done at the cygnus-ngsi Http listeners (in Flume jergon, sources) thanks to NGSIRestHandler. Once translated, the data (now, as NGSIEvent objects) is put into the internal channels for future consumption (see next section).
Mapping NGSIEvents to ArcGis
ArcGis stores data in it's own databases using it's own data organization, you can checkout this info Reading Feature Table details at Arcgis server, Such organization is exploited by NGSIArcgisFeatureTableSink each time a NGSIEvent is going to be persisted.
Arcgis feature tables must be provisioned before sending entities.
ArcGis databases naming conventions
Each entity type needs an url and an unique field to be persisted into the feature table.
NGSIArcgisFeatureTableSink composes each table's url with entitie's service and service path, to provide multiple tables access. The final url is composed of cygnus-ngsi.sinks.arcgis-sink.arcgis_service_url+fiware-service+fiware-servicepath.
Unique field is provided to allow NGSIArcgisFeatureTableSink to update existant entities. NGSI entity type will be used as unique field name. This means that a feature named type in the Feature Table cannot be filled in by the sink. If Feature Table needs to persist the value of entity type it has to be in a field different than type.
All this parameters, can be customized using Context Broker custom notifications (preferred) or Cygnus mapping capabilities.
Assuming that the feature table's url: https://arcgis.com/{hash}/arcgis/rest/services/vehicles/cars and feature table definition is:
Fields:
objectid ( type: esriFieldTypeOID, alias: objectid, editable: false, nullable: false, defaultValue: null, modelName: objectid )
licensePlate ( type: esriFieldTypeString, alias: licensePlate, editable: true, nullable: true, length: 255, defaultValue: null, modelName: licensePlate )
speed ( type: esriFieldTypeDouble, alias: speed, editable: true, nullable: true, defaultValue: null, modelName: speed )
oilLevel ( type: esriFieldTypeDouble, alias: oilLevel, editable: true, nullable: true, defaultValue: null, modelName: oilLevel )
Where licensePlate is the unique field to store the entityId value from the Context Broker.
Let's see both configuration options:
Using Context Broker custom notifications (preferred)
CB custom notifications is the preferred option because of its simplicity and not having to manage configuration (name-mappings) files in the server.
Entity data in CB:
service = vehicles
service-path = /4wheels
entity-type = Car
If the Feature table for type "Car" is https://arcgis.com/{hash}/arcgis/rest/services/vehicles/cars, the subscription with custom notif would be:
{
"description": "Subs arcgis",
"status": "active",
"subject": {
"entities": [
{
"idPattern": ".*",
"type": "Car"
}
],
"condition": {
"attrs": [
"speed",
"oilLevel"
]
}
},
"notification": {
"attrs": [
"speed",
"oilLevel"
],
"onlyChangedAttrs": false,
"attrsFormat": "normalized",
"httpCustom": {
"url": "http://iot-cygnus:<source_arcgis_port>/notify",
"ngsi": {
"type": "licensePlate"
},
"headers": {
"fiware-service": "vehicles",
"fiware-servicepath": "/cars"
}
}
}
}
Note that to avoid using the name mappings to modify the unique field value of type attribute, it is required the use of ngsi patching functionality.
result
Feature table url: https://arcgis.com/{hash}/arcgis/rest/services/vehicles/cars
Table's unique field: licensePlate
Using Cygnus name mappings
Agent.conf file:
agent.arcgis-sink.arcgis_service_url = https://arcgis.com/{hash}/arcgis/rest/services
agent.arcgis-sink.enable_name_mappings = true
Entity data:
service = vehicles
service-path = /4wheels
entity-type = Car
The name mappings configuration would be:
{
"serviceMappings": [{
"originalService": "vehicles",
"servicePathMappings": [{
"originalServicePath": "/4wheels",
"newServicePath": "/cars",
"entityMappings": [{
"originalEntityType": "Car",
"newEntityType": "licensePlate",
"originalEntityId": "^.*"
}
]
}
]
}
]
}
result
Feature table url: https://arcgis.com/{hash}/arcgis/rest/services/vehicles/cars
Table's unique field: licensePlate
Administration guide
Configuration
NGSIArcgisFeatureTableSink is configured through the following parameters:
| Parameter | Mandatory | Default value | Comments |
|---|---|---|---|
| type | yes | N/A | Must be com.telefonica.iot.cygnus.sinks.NGSIArcgisFeatureTableSink |
| channel | yes | N/A | |
| enable_encoding | no | false | true or false, true applies the new encoding, false applies the old encoding. |
| enable_name_mappings | no | false | true or false. Check this link for more details. |
| arcgis_service_url | yes | N/A | https://{url_host}/{id_arcgis}/arcgis/rest/services |
| arcgis_username | yes | N/A | |
| arcgis_password | yes | N/A | |
| arcgis_gettoken_url | yes | N/A | https://{url_host}/sharing/generateToken |
| arcgis_maxBatchSize | no | 10 | Number of events accumulated before persistence. |
| batch_timeout | no | 30 | Number of seconds the batch will be building before it is persisted as it is. |
| batch_ttl | no | 10 | Number of retries when a batch cannot be persisted. Use 0 for no retries, -1 for infinite retries. Please, consider an infinite TTL (even a very large one) may consume all the sink's channel capacity very quickly. |
A configuration example could be:
cygnus-ngsi.sinks = arcgis-sink
cygnus-ngsi.channels = arcgis-channel
...
cygnus-ngsi.sinks.arcgis-sink.type = com.telefonica.iot.cygnus.sinks.NGSIArcgisFeatureTableSink
cygnus-ngsi.sinks.arcgis-sink.channel = arcgis-channel
cygnus-ngsi.sinks.arcgis-sink.enable_name_mappings = true
cygnus-ngsi.sinks.arcgis-sink.arcgis_service_url = https://arcgis.com/UsuarioArcgis/arcgis/rest/services
cygnus-ngsi.sinks.arcgis-sink.arcgis_username = myuser
cygnus-ngsi.sinks.arcgis-sink.arcgis_password = mypassword
cygnus-ngsi.sinks.arcgis-sink.arcgis_gettoken_url = https://arcgis.com/sharing/generateToken
cygnus-ngsi.sinks.arcgis-sink.arcgis_maxBatchSize = 10
Important notes
About batching
As explained in the programmers guide, NGSIArcgisFeatureTableSink extends NGSISink, which provides a built-in mechanism for collecting events from the internal Flume channel. This mechanism allows extending classes have only to deal with the persistence details of such a batch of events in the final backend.
What is important regarding the batch mechanism is it largely increases the performance of the sink, because the number of writes is dramatically reduced. Let's see an example, let's assume a batch of 10 NGSIEvents. In the best case, all these events regard to the same type of entity, which means all the data within them will be persisted in the same ArcGis layer. If processing the events one by one, we would need 10 inserts into ArcGis; nevertheless, in this example only one insert is required. Obviously, not all the events will always regard to the same unique type of entity, and many entities may be involved within a batch.
The batch mechanism adds an accumulation timeout to prevent the sink stays in an eternal state of batch building when no new data arrives. If such a timeout is reached, then the batch is persisted as it is.
Regarding the retries of not persisted batches, a couple of parameters is used. On the one hand, a Time-To-Live (TTL) is used, specifing the number of retries Cygnus will do before definitely dropping the event. On the other hand, a list of retry intervals can be configured. Such a list defines the first retry interval, then se second retry interval, and so on; if the TTL is greater than the length of the list, then the last retry interval is repeated as many times as necessary.
By default, NGSIArcgisFeatureTableSink has a configured batch size and batch accumulation timeout of 1 and 30 seconds, respectively. Nevertheless, as explained above, it is highly recommended to increase at least the batch size for performance purposes. Which are the optimal values? The size of the batch it is closely related to the transaction size of the channel the events are got from (it has no sense the first one is greater then the second one), and it depends on the number of estimated sub-batches as well. The accumulation timeout will depend on how often you want to see new data in the final storage. A deeper discussion on the batches of events and their appropriate sizing may be found in the performance document.
Connections to cygnus-ngsi.sinks.arcgis-sink.arcgis_service_url and cygnus-ngsi.sinks.arcgis-sink.arcgis_service_url are done without check certificate validation of Java SSL Connections.
About case-sensitivity
FIXME #2320. Currently Arcgis sink is case sensitive with the attributes to persist in the Feature Table although arcgis is not case sensitive. This behaviour requires the use of name-mappings to match the case letters of the attribute's definition in the Feature Table.
For instance, if we have the following feature table definition for the "Car" entity type:
Fields:
objectid ( type: esriFieldTypeOID, alias: objectid, editable: false, nullable: false, defaultValue: null, modelName: objectid )
licensePlate ( type: esriFieldTypeString, alias: licensePlate, editable: true, nullable: true, length: 255, defaultValue: null, modelName: licensePlate )
speed ( type: esriFieldTypeDouble, alias: speed, editable: true, nullable: true, defaultValue: null, modelName: speed )
oillevel ( type: esriFieldTypeDouble, alias: oillevel, editable: true, nullable: true, defaultValue: null, modelName: oillevel )
Note that field oillevel is completely in lower case.
And the model definition of the Car is:
{
"id": "car1",
"type": "Car",
"location": {
"type": "geo:json",
"value": {
"coordinates": [
-0.350062,
40.054448
],
"type": "Point"
}
},
"speed": {
"type": "Number",
"value": 112.9
},
"oilLevel": {
"type": "Number",
"value": 74.6
}
}
With attribute oilLevel in camelCase format.
The name mappings required to persist the attributes is:
{
"serviceMappings": [{
"originalService": "vehicles",
"servicePathMappings": [{
"originalServicePath": "/4wheels",
"newServicePath": "/cars",
"entityMappings": [{
"originalEntityType": "Car",
"newEntityType": "licensePlate",
"originalEntityId": "^.*"
"attributeMappings": [{
"originalAttributeName": "oilLevel",
"newAttributeName": "oillevel",
"originalAttributeType": "^.*"
},
...
]
}
]
}
]
}
]
}
Note that speed attribute is not required in the name mappings file as it match, including case-sensitivity, with the field in the Feature Layer.
About Arcgis data types
- esriFieldTypeDate
From https://doc.arcgis.com/en/data-pipelines/latest/process/output-feature-layer.htm
... Date fields are stored in feature layers using the format milliseconds from epoch and the coordinated universal time (UTC) time zone. The values will be displayed differently depending on where you are viewing the data. For example, querying the feature service REST end point will return values in milliseconds from epoch, such as 1667411518878....
So, all NGSI DateTime are persisted like a esriFieldTypeDate field in the Feature Layer, because arcgis sink transform NGSI DateTime into the milliseconds from epoch.
Programmers guide
NGSIArcgisFeatureTableSink class
As any other NGSI-like sink, NGSIArcgisFeatureTableSink extends the base NGSISink. The methods that are extended are:
void persistBatch(NGSIBatch batch) throws CygnusBadConfiguration,
CygnusBadContextData, CygnusRuntimeError, CygnusPersistenceError;
A Batch contains a set of NGSIEvent objects, which are the result of parsing the notified context data events. Data within the batch is classified by destination, and in the end, a destination specifies the ArcGis where the data is going to be persisted. Thus, each destination is iterated in order to compose a per-destination data string to be persisted thanks to any Arcgis implementation.
public void start();
This must be done at the start() method and not in the constructor since the invoking sequence is NGSIArcgisFeatureTableSink() (contructor), configure() and start().
public void configure(Context);
A complete configuration as the described above is read from the given Context instance.
Authentication and authorization
Current implementation of NGSIArcgisFeatureTableSink relies on the username and password credentials created at the ArcGis endpoint.