Ingest JSON Array data into Azure Data Explorer (ADX) using .NET SDK & Azure Functions

Azure Data Explorer (ADX) is a fast and highly scalable data analytics service provided by Microsoft Azure. It is designed to help users quickly ingest, analyze, and visualize large volumes of structured and unstructured data from various sources. In this blog post, we will walk through the process of ingesting JSON data into ADX using Azure Functions and .NET SDK.

Ingesting Data into Azure Data Explorer:

There are several ways to ingest data into ADX, including batch ingestion, streaming ingestion, and continuous data ingestion. In our example, we will be using the .NET SDK in Azure Functions to ingest JSON data into Azure Data Explorer.

Setting up the Azure Data Explorer Cluster and Database:

1. Create an Azure Data Explorer cluster in the Azure portal.
2. Create a database within the cluster. In our example, we will name the database “test_database”.

Creating a Table and JSON Mapping in Azure Data Explorer:

1. Create a table in the database with the required columns to store the JSON data.
2. Create a JSON mapping in ADX. This mapping will determine which JSON property goes into which ADX table column.

Refer to https://learn.microsoft.com/en-us/azure/data-explorer/ingest-json-formats?tabs=kusto-query-language

Creating an Azure Function to Ingest JSON Data:

1. Create an Azure Function using .NET 6.
2. Install the necessary NuGet packages (Kusto.Data and Kusto.Ingest).
3. Create a class named “IngestJson” to handle the ADX related code.
4. In this class, create a method to ingest the JSON payload into the events table.

    public class IngestJSON
    {
        private string IngestURI;

        private string DatabaseName;

        private string TableName;

        public IngestJSON(string ingestURI, string databaseName, string tableName)
        {
            IngestURI = ingestURI;
            DatabaseName = databaseName;
            TableName = tableName;
        }

        public async Task IngestJsonPayloadAsync(string jsonPayload)
        {
     
            byte[] jsonPayloadBytes = Encoding.ASCII.GetBytes(jsonPayload);
            long requestBodySize = jsonPayloadBytes.LongLength;

            var sourceOptions = new StreamSourceOptions
            {
                Size = requestBodySize,
                SourceId = Guid.NewGuid()
            };

            var prop = CreateIngestionProperties(DataSourceFormat.multijson, "FlatEventMapping");

            try
            {
                KustoConnectionStringBuilder connectionStringBuilder = new KustoConnectionStringBuilder(IngestURI).WithAadSystemManagedIdentity();
                using (var client = KustoIngestFactory.CreateQueuedIngestClient(connectionStringBuilder))
                {
                    var result = await client.IngestFromStreamAsync(new MemoryStream(jsonPayloadBytes), prop, sourceOptions);
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        private KustoIngestionProperties CreateIngestionProperties(DataSourceFormat dataFormat, string mappingName)
        {
            var kustoIngestionProperties = new KustoIngestionProperties()
            {
                DatabaseName = DatabaseName,
                TableName = TableName,
                IngestionMapping = new IngestionMapping() { IngestionMappingReference = mappingName },
                Format = dataFormat,
            };

            // Set the operation timeout
            kustoIngestionProperties.AdditionalProperties.Add("OperationTimeoutMs", "60000");

            return kustoIngestionProperties;
        }
    }

Testing the Azure Function:

1. Deploy the Azure Function to Azure.
2. Enable System Managed Identity for the function.
3. Add the necessary application settings in the Azure Function.
4. Test the function by sending a JSON payload using a tool like Postman.

Handling JSON Arrays:

To handle JSON arrays and store each object in the array as a separate row in the events table, follow these steps:

1. Create a staging table (raw_events) to store the entire JSON payload.
2. Create a function in ADX to expand the raw_events data and project them as individual rows.
3. Create a table mapping for the raw_events table.
4. Create an update policy that triggers on the insertion of a new item into the raw_events table. This policy will execute the function that unpacks the JSON array into individual rows and inserts them into the target table (events).

In conclusion, this blog post has demonstrated how to ingest JSON data into Azure Data Explorer using Azure Functions and .NET SDK. By following the steps outlined above, you can easily ingest, analyze, and visualize JSON data from various sources in Azure Data Explorer.