Data ingestion

• 1: AVRO Mapping
• 2: CSV Mapping
• 3: Ingestion mappings
• 4: JSON Mapping
• 5: ORC Mapping
• 6: Parquet Mapping
• 7: Queued ingestion commands use case
• 8: Queued ingestion overview commands
• 9: W3CLOGFILE Mapping
• 10: .ingest inline command (push)
• 11: .show data operations
• 12: Data formats supported for ingestion
• 13: Data ingestion properties
• 14: Ingest from query
• 14.1: .cancel operation command
• 14.2: Kusto query ingestion (set, append, replace)
• 15: Kusto.ingest into command (pull data from storage)
• 16: Streaming ingestion
• 16.1: Clearing cached schema for streaming ingestion
• 16.2: Streaming ingestion and schema changes

1 - AVRO Mapping

Use AVRO mapping to map incoming data to columns inside tables when your ingestion source file is in AVRO format.

Each AVRO mapping element must contain either of the following optional properties:

Examples

JSON serialization

The following example mapping is serialized as a JSON string when provided as part of the .ingest management command.

[
  {"Column": "event_timestamp", "Properties": {"Field": "Timestamp"}},
  {"Column": "event_name",      "Properties": {"Field": "Name"}},
  {"Column": "event_type",      "Properties": {"Field": "Type"}},
  {"Column": "event_time",      "Properties": {"Field": "Timestamp", "Transform": "DateTimeFromUnixMilliseconds"}},
  {"Column": "ingestion_time",  "Properties": {"ConstValue": "2021-01-01T10:32:00"}},
  {"Column": "full_record",     "Properties": {"Path": "$"}}
]

Here the serialized JSON mapping is included in the context of the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
  with
  (
      format = "AVRO",
      ingestionMapping =
      ```
      [
        {"Column": "column_a", "Properties": {"Field": "Field1"}},
        {"Column": "column_b", "Properties": {"Field": "$.[\'Field name with space\']"}}
      ]
      ```
  )

Precreated mapping

When the mapping is precreated, reference the mapping by name in the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="AVRO",
        ingestionMappingReference = "Mapping_Name"
    )

Identity mapping

Use AVRO mapping during ingestion without defining a mapping schema (see identity mapping).

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="AVRO"
    )

Related content

• Use the avrotize k2a tool to create an Avro schema.

2 - CSV Mapping

Use CSV mapping to map incoming data to columns inside tables when your ingestion source file is any of the following delimiter-separated tabular formats: CSV, TSV, PSV, SCSV, SOHsv, TXT and RAW. For more information, see supported data formats.

Each CSV mapping element must contain either of the following optional properties:

Examples

[
  {"Column": "event_time", "Properties": {"Ordinal": "0"}},
  {"Column": "event_name", "Properties": {"Ordinal": "1"}},
  {"Column": "event_type", "Properties": {"Ordinal": "2"}},
  {"Column": "ingestion_time", "Properties": {"ConstValue": "2023-01-01T10:32:00"}}
  {"Column": "source_location", "Properties": {"Transform": "SourceLocation"}}
]

The mapping above is serialized as a JSON string when it’s provided as part of the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="csv",
        ingestionMapping =
        ```
        [
            {"Column": "event_time", "Properties": {"Ordinal": "0"}},
            {"Column": "event_name", "Properties": {"Ordinal": "1"}},
            {"Column": "event_type", "Properties": {"Ordinal": "2"}},
            {"Column": "ingestion_time", "Properties": {"ConstValue": "2023-01-01T10:32:00"}},
            {"Column": "source_location", "Properties": {"Transform": "SourceLocation"}}
        ]
        ```
    )

Pre-created mapping

When the mapping is pre-created, reference the mapping by name in the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="csv",
        ingestionMappingReference = "MappingName"
    )

Identity mapping

Use CSV mapping during ingestion without defining a mapping schema (see identity mapping).

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="csv"
    )

3 - Ingestion mappings

Ingestion mappings are used during ingestion to map incoming data to columns inside tables.

Data Explorer supports different types of mappings, both row-oriented (CSV, JSON, AVRO and W3CLOGFILE), and column-oriented (Parquet and ORC).

Ingestion mappings can be defined in the ingest command, or can be precreated and referenced from the ingest command using ingestionMappingReference parameters. Ingestion is possible without specifying a mapping. For more information, see identity mapping.

Each element in the mapping list is constructed from three fields:

Supported mapping types

The following table defines mapping types to be used when ingesting or querying external data of a specific format.

Ingestion mapping examples

The following examples use the RawEvents table with the following schema:

.create table RawEvents (timestamp: datetime, deviceId: guid, messageId: guid, temperature: decimal, humidity: decimal) 

Simple mapping

The following example shows ingestion where the mapping is defined in the ingest command. The command ingests a JSON file from a URL into the RawEvents table. The mapping specifies the path to each field in the JSON file.

.ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') 
    with (
            format = "json",
            ingestionMapping =
            ```
            [ 
              {"column":"timestamp","Properties":{"path":"$.timestamp"}},
              {"column":"deviceId","Properties":{"path":"$.deviceId"}},
              {"column":"messageId","Properties":{"path":"$.messageId"}},
              {"column":"temperature","Properties":{"path":"$.temperature"}},
              {"column":"humidity","Properties":{"path":"$.humidity"}}
            ]
            ```
          )

Mapping with ingestionMappingReference

To map the same JSON file using a precreated mapping, create the RawEventMapping ingestion mapping reference with the following command:

.create table RawEvents ingestion json mapping 'RawEventMapping' 
  ```
  [ 
    {"column":"timestamp","Properties":{"path":"$.timestamp"}},
    {"column":"deviceId","Properties":{"path":"$.deviceId"}},
    {"column":"messageId","Properties":{"path":"$.messageId"}},
    {"column":"temperature","Properties":{"path":"$.temperature"}},
    {"column":"humidity","Properties":{"path":"$.humidity"}}
  ]
  ```

Ingest the JSON file using the RawEventMapping ingestion mapping reference with the following command:

.ingest into table RawEvents ('https://kustosamplefiles.blob.core.windows.net/jsonsamplefiles/simple.json') 
  with (
          format="json",
          ingestionMappingReference="RawEventMapping"
        )

Identity mapping

Ingestion is possible without specifying ingestionMapping or ingestionMappingReference properties. The data is mapped using an identity data mapping derived from the table’s schema. The table schema remains the same. format property should be specified. See ingestion formats.

Mapping transformations

Some of the data format mappings (Parquet, JSON, and AVRO) support simple and useful ingest-time transformations. Where the scenario requires more complex processing at ingest time, use Update policy, which allows defining lightweight processing using KQL expression.

Mapping transformation examples

DropMappedFields transformation:

Given the following JSON contents:

{
    "Time": "2012-01-15T10:45",
    "Props": {
        "EventName": "CustomEvent",
        "Revenue": 0.456
    }
}

The following data mapping maps entire Props object into dynamic column Props while excluding already mapped columns (Props.EventName is already mapped into column EventName, so it’s excluded).

[
    { "Column": "Time", "Properties": { "Path": "$.Time" } },
    { "Column": "EventName", "Properties": { "Path": "$.Props.EventName" } },
    { "Column": "Props", "Properties": { "Path": "$.Props", "Transform":"DropMappedFields" } },
]

The ingested data looks as follows:

BytesAsBase64 transformation

Given the following AVRO file contents:

{
    "Time": "2012-01-15T10:45",
    "Props": {
        "id": [227,131,34,92,28,91,65,72,134,138,9,133,51,45,104,52]
    }
}

The following data mapping maps the ID column twice, with and without the transformation.

[
    { "Column": "ID", "Properties": { "Path": "$.props.id" } },
    { "Column": "Base64EncodedId", "Properties": { "Path": "$.props.id", "Transform":"BytesAsBase64" } },
]

The ingested data looks as follows:

4 - JSON Mapping

Use JSON mapping to map incoming data to columns inside tables when your ingestion source file is in JSON format.

Each JSON mapping element must contain either of the following optional properties:

Examples

[
  {"Column": "event_timestamp", "Properties": {"Path": "$.Timestamp"}},
  {"Column": "event_name",      "Properties": {"Path": "$.Event.Name"}},
  {"Column": "event_type",      "Properties": {"Path": "$.Event.Type"}},
  {"Column": "source_uri",      "Properties": {"Transform": "SourceLocation"}},
  {"Column": "source_line",     "Properties": {"Transform": "SourceLineNumber"}},
  {"Column": "event_time",      "Properties": {"Path": "$.Timestamp", "Transform": "DateTimeFromUnixMilliseconds"}},
  {"Column": "ingestion_time",  "Properties": {"ConstValue": "2021-01-01T10:32:00"}},
  {"Column": "full_record",     "Properties": {"Path": "$"}}
]

The mapping above is serialized as a JSON string when it’s provided as part of the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
  with
  (
      format = "json",
      ingestionMapping =
      ```
      [
        {"Column": "column_a", "Properties": {"Path": "$.Obj.Property"}},
        {"Column": "column_b", "Properties": {"Path": "$.Property"}},
        {"Column": "custom_column", "Properties": {"Path": "$.[\'Property name with space\']"}}
      ]
      ```
  )

Pre-created mapping

When the mapping is pre-created, reference the mapping by name in the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="json",
        ingestionMappingReference = "Mapping_Name"
    )

Identity mapping

Use JSON mapping during ingestion without defining a mapping schema (see identity mapping).

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="json"
    )

Copying JSON mapping

You can copy JSON mapping of an existing table and create a new table with the same mapping using the following process:

1. Run the following command on the table whose mapping you want to copy:

   .show table TABLENAME ingestion json mappings
   | extend formatted_mapping = strcat("'",replace_string(Mapping, "'", "\\'"),"'")
   | project formatted_mapping
   

2. Use the output of the above command to create a new table with the same mapping:

   .create table TABLENAME ingestion json mapping "TABLENAME_Mapping" RESULT_OF_ABOVE_CMD
   

5 - ORC Mapping

Use ORC mapping to map incoming data to columns inside tables when your ingestion source file is in ORC format.

Each ORC mapping element must contain either of the following optional properties:

Examples

[
  {"Column": "event_timestamp", "Properties": {"Path": "$.Timestamp"}},
  {"Column": "event_name",      "Properties": {"Path": "$.Event.Name"}},
  {"Column": "event_type",      "Properties": {"Path": "$.Event.Type"}},
  {"Column": "event_time",      "Properties": {"Path": "$.Timestamp", "Transform": "DateTimeFromUnixMilliseconds"}},
  {"Column": "ingestion_time",  "Properties": {"ConstValue": "2021-01-01T10:32:00"}},
  {"Column": "full_record",     "Properties": {"Path": "$"}}
]

The mapping above is serialized as a JSON string when it’s provided as part of the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
  with
  (
      format = "orc",
      ingestionMapping =
      ```
      [
        {"Column": "column_a", "Properties": {"Path": "$.Field1"}},
        {"Column": "column_b", "Properties": {"Path": "$.[\'Field name with space\']"}}
      ]
      ```
  )

Pre-created mapping

When the mapping is pre-created, reference the mapping by name in the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="orc",
        ingestionMappingReference = "ORC_Mapping"
    )

Identity mapping

Use ORC mapping during ingestion without defining a mapping schema (see identity mapping).

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="orc"
    )

6 - Parquet Mapping

Use Parquet mapping to map incoming data to columns inside tables when your ingestion source file is in Parquet format.

Each Parquet mapping element must contain either of the following optional properties:

Parquet type conversions

Comprehensive support is provided for converting data types when you’re ingesting or querying data from a Parquet source.

The following table provides a mapping of Parquet field types, and the table column types they can be converted to. The first column lists the Parquet type, and the others show the table column types they can be converted to.

Examples

[
  {"Column": "event_timestamp", "Properties": {"Path": "$.Timestamp"}},
  {"Column": "event_name",      "Properties": {"Path": "$.Event.Name"}},
  {"Column": "event_type",      "Properties": {"Path": "$.Event.Type"}},
  {"Column": "event_time",      "Properties": {"Path": "$.Timestamp", "Transform": "DateTimeFromUnixMilliseconds"}},
  {"Column": "ingestion_time",  "Properties": {"ConstValue": "2021-01-01T10:32:00"}},
  {"Column": "full_record",     "Properties": {"Path": "$"}}
]

The mapping above is serialized as a JSON string when it’s provided as part of the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
  with
  (
    format = "parquet",
    ingestionMapping =
    ```
    [
      {"Column": "column_a", "Properties": {"Path": "$.Field1.Subfield"}},
      {"Column": "column_b", "Properties": {"Path": "$.[\'Field name with space\']"}},
    ]
    ```
  )

Pre-created mapping

When the mapping is pre-created, reference the mapping by name in the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
  with
  (
      format="parquet",
      ingestionMappingReference = "Mapping_Name"
  )

Identity mapping

Use Parquet mapping during ingestion without defining a mapping schema (see identity mapping).

.ingest into Table123 (@"source1", @"source2")
  with
  (
    format="parquet"
  )

7 - Queued ingestion commands use case

The queued ingestion commands allow you to ingest individual blobs by URL or ingest batches of data by listing folders or containers. This article walks through a common use case: fine-tuning the ingestion of historical data. You can use these commands to test how historical data is ingested and resolve any issues before performing full ingestion. The following tasks demonstrate how to use queued ingestion commands effectively:

• Ingest single blobs (Preview)
• List blobs in a folder
• Ingest folder
• Track ingestion status
• Filter queued files for ingestion
• Capture the creation time
• Ingest 20 files
• Track follow up ingestion status
• Perform your full ingestion
• Cancel ingestion

Ingest single blobs (Preview)

You can start by ingesting a single blob directly using its URL. Make sure to include a SAS token or use a managed identity to grant the service permission to access and download the blob.

.ingest-from-storage-queued into table database('TestDatabase').Logs
with (format='csv', enableTracking=true)
<|
'https://sample.blob.core.windows.net/sample/test_*csv?...'

Output

List blobs in a folder

To understand the historical data better, you list a maximum of 10 blobs from the Azure blob storage container.

.list blobs (
    "https://<BlobStorageLocation>/<FolderName>;managed_identity=system"
)
MaxFiles=10

Output

You can now verify if the blobs are the correct blobs to ingest.

Ingest folder

Next you queue 10 parquet files for ingestion into the Logs table in the TestDatabase database with tracking enabled for the ingestion.

.ingest-from-storage-queued into table database('TestDatabase').Logs
EnableTracking=true
with (format='parquet')
<|
    .list blobs (
        "https://<BlobStorageLocation>/<FolderName>;managed_identity=system"
    )
    MaxFiles=10

Output

The OperationInfo, which includes the IngestionOperationId, is then used to track the ingestion status.

Track ingestion status

You run the .show queued ingestion operations command to check whether the ingestion is complete or if there are any errors.

.show queued ingestion operations "00001111;11112222;00001111-aaaa-2222-bbbb-3333cccc4444"

Output

If the State isn’t Completed, you can run the .show queued ingestion operations again. Running it again allows you to monitor the increase in the number of ingested blobs until the State changes to Completed. You can also cancel the ingestion, if necessary.

Filter queued files for ingestion

After the results of the ingestion are examined, another attempt at listing blobs for ingestion is made. This time the parquet suffix is added to ensure that only parquet files are ingested.

.list blobs (
    "https://<BlobStorageLocation>/<FolderName>;managed_identity=system"
)
Suffix="parquet"
MaxFiles=10

Output

Capture the creation time

A path format is added to capture the creation time.

.list blobs (
    "https://<BlobStorageLocation>/<FolderName>;managed_identity=system"
)
Suffix="parquet"
MaxFiles=10
PathFormat=("output/03/Year=" datetime_pattern("yyyy'/Month='MM'/Day='dd", creationTime) "/")

Output

The CapturedVariables column dates match the dates specified in the BlobUri column.

Ingest 20 files

Now 20 files in parquet format are ingested from the Azure blob storage container, along with their creation time.

.ingest-from-storage-queued into table database('TestDatabase').Logs
EnableTracking=true
with (format='parquet')
<|
    .list blobs (
        "https://<BlobStorageLocation>/<FolderName>;managed_identity=system"
    )
    Suffix="parquet"
    MaxFiles=20
    PathFormat=("output/03/Year=" datetime_pattern("yyyy'/Month='MM'/Day='dd", creationTime) "/")

Output

The OperationInfo is then used to track the ingestion status.

Track follow up ingestion status

The .show queued ingestion operations command is run to check whether there are any issues with this ingestion.

.show queued ingestion operations "22223333;22223333;11110000-bbbb-2222-cccc-4444dddd5555"

Output

|IngestionOperationId|Started On |Last Updated On |State |Discovered |InProgress|Canceled|Ingested |Failed|Canceled |SampleFailedReasons|Database|Table| |–|–|–|–|–|–|–|–|–|–|–|–| |22223333;22223333;11110000-bbbb-2222-cccc-4444dddd5555 |2025-02-20 14:57:41.0000000 | | InProgress| 10 |10 |0 |0 |0 | |TestDatabase|Logs|

The .show extents command is run to check whether extents are created with an anterior date for data integrity and historical accuracy.

.show table Logs extents

The MinCreatedOn and MaxCreatedOn values should show the data creation time, rather than the data ingestion time. For more information about these returns, see .show extents.

You can cancel the ingestion, if necessary.

Perform your full ingestion

By running queued ingestion commands on a sample, you can discover problems your ingestion might encounter. Now that you fixed them, you’re ready to ingest all your historical data and wait for the full ingestion to complete.

Cancel ingestion

At any time during the ingestion process, you can cancel your queued ingestion.

.cancel queued ingestion operation '22223333;22223333;11110000-bbbb-2222-cccc-4444dddd5555'

Output

You can then roll back the ingestion, fix the issues, and rerun the ingestion.

Related content

• Queued ingestion overview
• Data formats supported for ingestion
• .show extents

8 - Queued ingestion overview commands

Queued ingestion commands allow you to ingest specific folders, or an entire container and manage the operations related to queued ingestion. You can also ingest multiple or individual blobs by URL and from a source file. The ingestion commands are useful for preparing and testing distinct ingestion scenarios before the final ingestion. Using them helps to ensure that fields, columns, partitioning, and other needs are handled properly during ingestion.

Management commands

Queued storage commands include:

• .ingest-from-storage-queued into command queues blobs for ingestion into a table.
• .show queued ingestion operations command shows the queued ingestion operations.
• .cancel queued ingestion operation command cancels a queued ingestion operation.
• .list blobs command lists the blobs for ingestion.

Related content

• Queued ingestion use case
• Data formats supported for ingestion

9 - W3CLOGFILE Mapping

Use W3CLOGFILE mapping to map incoming data to columns inside tables when your ingestion source file is in W3CLOGFILE format.

Each W3CLOGFILE mapping element must contain either of the following optional properties:

Examples

[
   {"Column": "Date",          "Properties": {"Field": "date"}},
   {"Column": "Time",          "Properties": {"Field": "time"}},
   {"Column": "IP",            "Properties": {"Field": "s-ip"}},
   {"Column": "ClientMethod",  "Properties": {"Field": "cs-method"}},
   {"Column": "ClientQuery",   "Properties": {"Field": "cs-uri-query"}},
   {"Column": "ServerPort",    "Properties": {"Field": "s-port"}},
   {"Column": "ClientIP",      "Properties": {"Field": "c-ip"}},
   {"Column": "UserAgent",     "Properties": {"Field": "cs(User-Agent)"}},
   {"Column": "Referer",       "Properties": {"Field": "cs(Referer)"}},
   {"Column": "Status",        "Properties": {"Field": "sc-status"}},
   {"Column": "ResponseBytes", "Properties": {"Field": "sc-bytes"}},
   {"Column": "RequestBytes",  "Properties": {"Field": "cs-bytes"}},
   {"Column": "TimeTaken",     "Properties": {"Field": "time-taken"}}
]

The mapping above is serialized as a JSON string when it’s provided as part of the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
  with
  (
      format = "w3clogfile",
      ingestionMapping =
      ```
      [
         {"Column": "column_a", "Properties": {"Field": "field1"}},
         {"Column": "column_b", "Properties": {"Field": "field2"}}
      ]
      ```
  )

Pre-created mapping

When the mapping is pre-created, reference the mapping by name in the .ingest management command.

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="w3clogfile",
        ingestionMappingReference = "Mapping_Name"
    )

Identity mapping

Use W3CLOGFILE mapping during ingestion without defining a mapping schema (see identity mapping).

.ingest into Table123 (@"source1", @"source2")
    with
    (
        format="w3clogfile"
    )

10 - .ingest inline command (push)

This command inserts data into a table by pushing the data included within the command to the table.

Permissions

You must have at least Table Ingestor permissions to run this command.

Syntax

.ingest inline into table TableName [with ( IngestionPropertyName = IngestionPropertyValue [, …] )] <| Data

.ingest inline into table TableName [with ( IngestionPropertyName = IngestionPropertyValue [, …] )] [ Data ]

Parameters

Returns

The result is a table with as many records as the number of generated data shards (“extents”). If no data shards are generated, a single record is returned with an empty (zero-valued) extent ID.

Examples

Ingest with <| syntax

The following command ingests data into a table Purchases with two columns: SKU (of type string) and Quantity (of type long).

.ingest inline into table Purchases <|
    Shoes,1000
    Wide Shoes,50
    "Coats black",20
    "Coats with ""quotes""",5

Ingest with bracket syntax

The following command ingests data into a table Logs with two columns: Date (of type datetime) and EventDetails (of type dynamic).

.ingest inline into table Logs
    [2015-01-01,"{""EventType"":""Read"", ""Count"":""12""}"]
    [2015-01-01,"{""EventType"":""Write"", ""EventValue"":""84""}"]

Related content

• Data formats supported for ingestion
• .ingest into (pull data from storage)
• Ingest from query

11 - .show data operations

Returns a table with data operations that reached a final state. Data operations are available for 30 days from when they ran.

Any operation that results in new extents (data shards) added to the system is considered a data operation.

Permissions

You must have Database Admin or Database Monitor permissions to see any data operations invoked on your database.

Any user can see their own data operations.

For more information, see Kusto role-based access control.

Syntax

.show data operations

Returns

This command returns a table with the following columns:

Example

The following example returns information about UpdatePolicy, BatchIngest, and SetOrAppend operations.

.show data operations

Output

12 - Data formats supported for ingestion

Data ingestion adds data to a table and makes it available for query. For all ingestion methods, other than ingest-from-query, the data must be in one of the supported formats. The following table lists and describes the formats that are supported for data ingestion.

To learn why ingestion might fail, see Ingestion failures. and Ingestion error codes in Azure Data Explorer.

For more information about ingesting data by using the json or multijson formats, see Ingest JSON formats.

Supported data compression formats

Compress blobs and files with these algorithms:

Indicate compression by appending the extension to the blob or file name.

For example:

• MyData.csv.zip indicates a blob or file formatted as CSV, compressed with zip (archive or single file).
• MyData.json.gz indicates a blob or file formatted as JSON, compressed with gzip.

Blob or file names that include only the compression extension (for example, MyData.zip) are also supported. In this case, specify the file format as an ingestion property because it can’t be inferred.

Related content

• Supported data formats
• Data ingestion properties
• Data ingestion

13 - Data ingestion properties

Data ingestion adds data to a table and makes it available for query. Add properties to the ingestion command after the with keyword.

Related content

• Supported data formats
• Supported data formats
• Data ingestion

14 - Ingest from query

14.1 - .cancel operation command

This command cancels a long-running ingest from query operation. This command is useful when the operation is taking too long and you would like to abort it while running.

The cancel operation command isn’t guaranteed to succeed. The output of the .cancel operation command indicates whether or not cancellation was successful.

Syntax

.cancel operation OperationId [with ( reason = ReasonPhrase )]

Parameters

Returns

Example

.cancel operation 078b2641-f10d-4694-96f8-1ee2b75dda48 with(Reason="Command canceled by me")

14.2 - Kusto query ingestion (set, append, replace)

These commands execute a query or a management command and ingest the results of the query into a table. The difference between these commands is how they treat existing or nonexistent tables and data.

To cancel an ingest from query command, see cancel operation.

Permissions

To perform different actions on a table, you need specific permissions:

• To add rows to an existing table using the .append command, you need a minimum of Table Ingestor permissions.
• To create a new table using the various .set commands, you need a minimum of Database User permissions.
• To replace rows in an existing table using the .set-or-replace command, you need a minimum of Table Admin permissions.

For more information on permissions, see Kusto role-based access control.

Syntax

(.set | .append | .set-or-append | .set-or-replace) [async] tableName [with (propertyName = propertyValue [, …])] <| queryOrCommand

Parameters

Performance tips

• Set the distributed property to true if the amount of data produced by the query is large, exceeds one gigabyte (GB), and doesn’t require serialization. Then, multiple nodes can produce output in parallel. Don’t use this flag when query results are small, since it might needlessly generate many small data shards.
• Data ingestion is a resource-intensive operation that might affect concurrent activities on the database, including running queries. Avoid running too many ingestion commands at the same time.
• Limit the data for ingestion to less than one GB per ingestion operation. If necessary, use multiple ingestion commands.

Supported ingestion properties

Schema considerations

• .set-or-replace preserves the schema unless one of extend_schema or recreate_schema ingestion properties is set to true.
• .set-or-append and .append commands preserve the schema unless the extend_schema ingestion property is set to true.
• Matching the result set schema to that of the target table is based on the column types. There’s no matching of column names. Make sure that the query result schema columns are in the same order as the table, otherwise data is ingested into the wrong columns.

Character limitation

The command fails if the query generates an entity name with the $ character. The entity names must comply with the naming rules, so the $ character must be removed for the ingest command to succeed.

For example, in the following query, the search operator generates a column $table. To store the query results, use project-rename to rename the column.

.set Texas <| search State has 'Texas' | project-rename tableName=$table

Returns

Returns information on the extents created because of the .set or .append command.

Examples

Create and update table from query source

The following query creates the :::no-loc text=“RecentErrors”::: table with the same schema as :::no-loc text=“LogsTable”:::. It updates :::no-loc text=“RecentErrors”::: with all error logs from :::no-loc text=“LogsTable”::: over the last hour.

.set RecentErrors <|
   LogsTable
   | where Level == "Error" and Timestamp > now() - time(1h)

Create and update table from query source using the distributed flag

The following example creates a new table called OldExtents in the database, asynchronously. The dataset is expected to be bigger than one GB (more than ~one million rows) so the distributed flag is used. It updates OldExtents with ExtentId entries from the MyExtents table that were created more than 30 days ago.

.set async OldExtents with(distributed=true) <|
   MyExtents 
   | where CreatedOn < now() - time(30d)
   | project ExtentId

Append data to table

The following example filters ExtentId entries in the MyExtents table that were created more than 30 days ago and appends the entries to the OldExtents table with associated tags.

.append OldExtents with(tags='["TagA","TagB"]') <| 
   MyExtents 
   | where CreatedOn < now() - time(30d) 
   | project ExtentId

Create or append a table with possibly existing tagged data

The following example either appends to or creates the OldExtents table asynchronously. It filters ExtentId entries in the MyExtents table that were created more than 30 days ago and specifies the tags to append to the new extents with ingest-by:myTag. The ingestIfNotExists parameter ensures that the ingestion only occurs if the data doesn’t already exist in the table with the specified tag.

.set-or-append async OldExtents with(tags='["ingest-by:myTag"]', ingestIfNotExists='["myTag"]') <|
   MyExtents
   | where CreatedOn < now() - time(30d)
   | project ExtentId

Create table or replace data with associated data

The following query replaces the data in the OldExtents table, or creates the table if it doesn’t already exist, with ExtentId entries in the MyExtents table that were created more than 30 days ago. Tag the new extent with ingest-by:myTag if the data doesn’t already exist in the table with the specified tag.

.set-or-replace async OldExtents with(tags='["ingest-by:myTag"]', ingestIfNotExists='["myTag"]') <| 
   MyExtents 
   | where CreatedOn < now() - time(30d) 
   | project ExtentId

Append data with associated data

The following example appends data to the OldExtents table asynchronously, using ExtentId entries from the MyExtents table that were created more than 30 days ago. It sets a specific creation time for the new extents.

.append async OldExtents with(creationTime='2017-02-13T11:09:36.7992775Z') <| 
   MyExtents 
   | where CreatedOn < now() - time(30d) 
   | project ExtentId     

Sample output

The following is a sample of the type of output you may see from your queries.

Related content

• Data formats supported for ingestion
• Inline ingestion
• Ingest from storage

15 - Kusto.ingest into command (pull data from storage)

The .ingest into command ingests data into a table by “pulling” the data from one or more cloud storage files. For example, the command can retrieve 1,000 CSV-formatted blobs from Azure Blob Storage, parse them, and ingest them together into a single target table. Data is appended to the table without affecting existing records, and without modifying the table’s schema.

Permissions

You must have at least Table Ingestor permissions to run this command.

Syntax

.ingest [async] into table TableName SourceDataLocator [with ( IngestionPropertyName = IngestionPropertyValue [, …] )]

Parameters

Authentication and authorization

Each storage connection string indicates the authorization method to use for access to the storage. Depending on the authorization method, the principal might need to be granted permissions on the external storage to perform the ingestion.

The following table lists the supported authentication methods and the permissions needed for ingesting data from external storage.

Returns

The result of the command is a table with as many records as there are data shards (“extents”) generated by the command. If no data shards were generated, a single record is returned with an empty (zero-valued) extent ID.

Examples

Azure Blob Storage with shared access signature

The following example instructs your database to read two blobs from Azure Blob Storage as CSV files, and ingest their contents into table T. The ... represents an Azure Storage shared access signature (SAS) which gives read access to each blob. Obfuscated strings (the h in front of the string values) are used to ensure that the SAS is never recorded.

.ingest into table T (
    h'https://contoso.blob.core.windows.net/container/file1.csv?...',
    h'https://contoso.blob.core.windows.net/container/file2.csv?...'
)

Azure Blob Storage with managed identity

The following example shows how to read a CSV file from Azure Blob Storage and ingest its contents into table T using managed identity authentication. Authentication uses the managed identity ID (object ID) assigned to the Azure Blob Storage in Azure. For more information, see Create a managed identity for storage containers.

.ingest into table T ('https://StorageAccount.blob.core.windows.net/Container/file.csv;managed_identity=802bada6-4d21-44b2-9d15-e66b29e4d63e')

Azure Data Lake Storage Gen 2

The following example is for ingesting data from Azure Data Lake Storage Gen 2 (ADLSv2). The credentials used here (...) are the storage account credentials (shared key), and we use string obfuscation only for the secret part of the connection string.

.ingest into table T (
  'abfss://myfilesystem@contoso.dfs.core.windows.net/path/to/file1.csv;...'
)

Azure Data Lake Storage

The following example ingests a single file from Azure Data Lake Storage (ADLS). It uses the user’s credentials to access ADLS (so there’s no need to treat the storage URI as containing a secret). It also shows how to specify ingestion properties.

.ingest into table T ('adl://contoso.azuredatalakestore.net/Path/To/File/file1.ext;impersonate')
  with (format='csv')

Amazon S3 with an access key

The following example ingests a single file from Amazon S3 using an access key ID and a secret access key.

.ingest into table T ('https://bucketname.s3.us-east-1.amazonaws.com/path/to/file.csv;AwsCredentials=AKIAIOSFODNN7EXAMPLE,wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY')
  with (format='csv')

Amazon S3 with a presigned URL

The following example ingests a single file from Amazon S3 using a preSigned URL.

  with (format='csv')

Related content

• Data formats supported for ingestion
• Inline ingestion
• Ingest from query (.set, .append, .set-or-append, .set-or-replace)
• .show ingestion failures command
• .show ingestion mapping

16 - Streaming ingestion

16.1 - Clearing cached schema for streaming ingestion

Nodes cache schema of the databases that receive data via streaming ingestion. This process optimizes performance and utilization of resources, but can cause propagation delays when the schema change.

Clear the cache to guarantee that subsequent streaming ingestion requests incorporate database or table schema changes. For more information, see Streaming ingestion and schema changes.

Permissions

You must have at least Database Ingestor permissions to run this command.

Syntax

.clear table TableName cache streamingingestion schema

.clear database cache streamingingestion schema

Parameters

Returns

This command returns a table with the following columns:

Example

.clear database cache streamingingestion schema

.clear table T1 cache streamingingestion schema

16.2 - Streaming ingestion and schema changes

Cluster nodes cache the schema of databases that get data through streaming ingestion, boosting performance and resource use. However, when there are schema changes, it can lead to delays in updates.

Eventhouse nodes cache the schema of databases that get data through streaming ingestion, boosting performance and resource use. However, when there are schema changes, it can lead to delays in updates.

If schema changes and streaming ingestion aren’t synchronized, you can encounter failures like schema-related errors or incomplete and distorted data in the table.

This article outlines typical schema changes and provides guidance on avoiding problems with streaming ingestion during these changes.

Schema changes

The following list covers key examples of schema changes:

• Creation of tables
• Deletion of tables
• Adding a column to a table
• Removing a column from a table
• Retyping the columns of a table
• Renaming the columns of a table
• Adding precreated ingestion mappings
• Removing precreated ingestion mappings
• Adding, removing, or altering policies

Coordinate schema changes with streaming ingestion

The schema cache is kept while the database is online. If there are schema changes, the system automatically refreshes the cache, but this refresh can take several minutes. If you rely on the automatic refresh, you can experience uncoordinated ingestion failures.

You can reduce the effects of propagation delay by explicitly clearing the schema cache on the nodes. If the streaming ingestion flow and schema changes are coordinated, you can completely eliminate failures and their associated data distortion.

To coordinate the streaming ingestion flow with schema changes:

1. Suspend streaming ingestion.
2. Wait until all outstanding streaming ingestion requests are complete.
3. Do schema changes.
4. Issue one or several .clear cache streaming ingestion schema commands.
   • Repeat until successful and all rows in the command output indicate success
5. Resume streaming ingestion.