Skip to main content

Direct ingestion to ADX

A standard flow foresees data flow as below:

IoT Hub message / a file in saTwinMapper Storage Account -> Twin Mapper -> Azure Digital Twin -> Azure Data Explorer

There are certain Products of Generic CSV use case, which don't need the Azure Digital Twin instance updates. Using it in this case would make the data ingestion way slower and most likely load the ADT instance to its Quota Limits.

Therefore for a particular Product setting where the type is set to IMPORT_JOB and the Azure Digital Twin is set to the one starting with gencsv, an additional option will appear in the Product configuration. By default Update Digital Twins is set to true - as a standard flow foresees.

If flag Update Digital Twins is off, Twin Mapper will send data directly to the ADX, avoiding sending any updates to ADT.

Twin Mapper will process the input file in the same way as any other Generic CSV file (it will apply the Product Mapping). Output file will be in TSV format (like CSV but delimiter is a tabulator character '\t').

For this flow:

  • original file will be saved in processed or failed folder (same as for Bulk Import)
  • the processed file will be saved in processed_adx folder
  • data will be sent to ADX as a stream

RTCMA​

Similar case if for RTMCA product, for which the upload happens every 5 min, therefore for the same Quota Limits concerns, we added an option to skip Digital Twins in the flow and ingest directly to ADX. Table for RTCMA direct load is called rtcmaDirectLoadEvents. The flow works in exactly the same way - the same table mapping and upload and processed file directories are same respectfully as for RTCMA case.

In order to make this flow work, the new table in ADX has to be created on each environment.

  • generic CSV table name: genCsvDirectLoadEvents
  • RTCMA table name: rtcmaDirectLoadEvents
// create a table with the same schema as ADT tables
.create table <table-name> (TimeStamp: datetime, ServiceId: string, Id: string, ModelId: string, Key: string, Value: dynamic, RelationshipTarget: string, RelationshipId: string);
// create mapping for the table
.create table <table-name> ingestion csv mapping "csvMappingRef"
```
[
  {"Column": "TimeStamp", "DataType":"datetime", "Properties": {"Ordinal": "0"}},
  {"Column": "ServiceId", "DataType":"string", "Properties": {"Ordinal": "1"}},
  {"Column": "Id", "DataType":"string", "Properties": {"Ordinal": "2"}},
  {"Column": "ModelId", "DataType":"string", "Properties": {"Ordinal": "3"}},
  {"Column": "Key", "DataType":"string", "Properties": {"Ordinal": "4"}},
  {"Column": "Value", "DataType":"dynamic", "Properties": {"Ordinal": "5"}}
]
```;
// add twin mapper as ingestor to the table
.add table <table-name> ingestors ('aadapp=<twin mapper managed identity ID>');