Create an app to get data using the managed streaming ingestion client

The code sample includes a service function PrintResultAsValueList() for printing query results.

Add the Kusto libraries using the following commands:

dotnet add package Microsoft.Azure.Kusto.Data
dotnet add package Microsoft.Azure.Kusto.ingest

using System;
using Kusto.Data;
using Kusto.Data.Net.Client;
using Kusto.Ingest;
using Kusto.Data.Common;
using Microsoft.Identity.Client;
using System.Data;
using System.Text;

class Program
{
    static void Main(string[] args)
    {
        var tableName = "MyStormEvents";
        var clusterUrl = "<KustoClusterQueryURI>";
        var ingestionUrl = "<KustoClusterQueryIngestURI>";
        var databaseName = "<databaseName>";

        var clusterKcsb = new KustoConnectionStringBuilder(clusterUrl).WithAadUserPromptAuthentication();
        var ingestionKcsb = new KustoConnectionStringBuilder(ingestionUrl).WithAadUserPromptAuthentication();

        using (var kustoClient = KustoClientFactory.CreateCslQueryProvider(clusterKcsb))
        using (var ingestClient = KustoIngestFactory.CreateManagedStreamingIngestClient(clusterKcsb, ingestionKcsb))
        {          
            Console.WriteLine("Number of rows in " + tableName);
            var queryProvider = KustoClientFactory.CreateCslQueryProvider(clusterKcsb);
            var result = kustoClient.ExecuteQuery(databaseName, tableName + " | count", new ClientRequestProperties());
    
            PrintResultAsValueList(result);
        }
    }


    static void PrintResultAsValueList(IDataReader result)
    {
        var row=0;
        while (result.Read())
        {   
            row ++;
            Console.WriteLine("row:" + row.ToString() + "\t");
            for (int i = 0; i < result.FieldCount; i++)
            {
                Console.WriteLine("\t" + result.GetName(i) + " - " + result.GetValue(i));
            }
            Console.WriteLine();
        }
    }
}

Use the IngestFromStorageAsync method to ingest the stormevents.csv file.

Copy stormevents.csv file to the same location as your script. Since our input is a CSV file, use Format = DataSourceFormat.csv in the ingestion properties.

Add and ingestion section using the following lines to the end of Main().

var ingestProperties = new KustoIngestionProperties(databaseName, tableName) 
    {
        Format = DataSourceFormat.csv
    };
//Ingestion section
Console.WriteLine("Ingesting data from a file");
ingestClient.IngestFromStorageAsync(".\\stormevents.csv", ingestProperties).Wait();

Let’s also query the new number of rows and the most recent row after the ingestion. Add the following lines after the ingestion command:

Console.WriteLine("Number of rows in " + tableName);
result = kustoClient.ExecuteQuery(databaseName, tableName + " | count", new ClientRequestProperties());
PrintResultAsValueList(result);

Console.WriteLine("Example line from " + tableName);
result = kustoClient.ExecuteQuery(databaseName, tableName + " | top 1 by EndTime", new ClientRequestProperties());
PrintResultAsValueList(result);

The code sample includes a service function print_result_as_value_list() for printing query results

import os
import io


from azure.kusto.data import KustoClient, KustoConnectionStringBuilder, ClientRequestProperties, DataFormat
from azure.kusto.ingest import QueuedIngestClient, FileDescriptor, StreamDescriptor, BlobDescriptor, IngestionProperties, ManagedStreamingIngestClient


def print_result_as_value_list(result):
    row_count = 1

    # create a list of columns
    cols = list((col.column_name for col in result.primary_results[0].columns))

    # print the values for each row
    for row in result.primary_results[0]:
        if row_count > 1:
            print("######################")

        print("row", row_count, ":")
        for col in cols:
            print("\t", col, "-", row[col])

def main():
    # Connect to the public access Help cluster
    file_path = os.curdir + "/stormevents.csv"
    cluster_url = "<KustoClusterQueryURI>"
    ingestion_url = "<KustoClusterQueryIngestURI>"
    database_name = "<databaseName>"
    table_name = "MyStormEvents"
    cluster_kcsb = KustoConnectionStringBuilder.with_interactive_login(cluster_url)
    ingestion_kcsb = KustoConnectionStringBuilder.with_interactive_login(ingestion_url)

    with KustoClient(cluster_kcsb) as kusto_client:
        with ManagedStreamingIngestClient(cluster_kcsb, ingestion_kcsb) as ingest_client:

            print("Number of rows in " + table_name)
            result = kusto_client.execute_query(database_name, table_name + " | count")
            print_result_as_value_list(result)

main()

Use the ingest_from_file() API to ingest the stormevents.csv file. Place the stormevents.csv file in the same location as your script. Since our input is a CSV file, use DataFormat.CSV in the ingestion properties.

Add and ingestion section using the following lines to the end of main().

# Ingestion section
print("Ingesting data from a file")
ingest_properties = IngestionProperties(database_name, table_name, DataFormat.CSV)
ingest_client.ingest_from_file(file_path, ingest_properties)

Let’s also query the new number of rows and the most recent row after the ingestion. Add the following lines after the ingestion command:

print("New number of rows in " + table_name)
result = kusto_client.execute_query(database_name, table_name + " | count")
print_result_as_value_list(result)

print("Example line from " + table_name)
result = kusto_client.execute_query(database_name, table_name + " | top 1 by EndTime")
print_result_as_value_list(result)

Run the script from the directory where the script and stormevents.csv are located. Alternatively, you can specify the full path to the file replacing file_path = os.curdir + "/stormevents.csv" with file_path = "<full path to stormevents.csv>"

The code sample includes a service function printResultAsValueList() for printing query results.

import { Client as KustoClient, KustoConnectionStringBuilder } from "azure-kusto-data";
import { InteractiveBrowserCredentialInBrowserOptions } from "@azure/identity";
import { ManagedStreamingIngestClient, BlobDescriptor, IngestionProperties, DataFormat } from 'azure-kusto-ingest';

const clusterUrl = "<KustoClusterQueryURI>";
const ingestionUrl ="<KustoClusterQueryIngestURI>";
const databaseName = "<databaseName>";

const tableName = "MyStormEvents"

    async function main() {
    const clusterKcsb = KustoConnectionStringBuilder.withUserPrompt(clusterUrl);
    const ingestionKcsb = KustoConnectionStringBuilder.withUserPrompt(ingestionUrl);

    const kustoClient = new KustoClient(clusterKcsb);
    const ingestClient = new ManagedStreamingIngestClient(clusterKcsb, ingestionKcsb);

    console.log(`Number of rows in ${tableName}`);
    let result = await kustoClient.executeQuery(databaseName, `${tableName} | count`);
    printResultAsValueList(result);
}

function printResultAsValueList(result: any) {
    let rowCount = 1;

    // Create a list of columns
    const cols = result.primaryResults[0].columns.map((col: any) => col.name);
    // Print the values for each row
    for (const row of result.primaryResults) {
        if (rowCount > 1) {
            console.log('######################');
        }

        console.log(`row ${rowCount}:`);
        for (const col of cols) {
            const jsonObject = JSON.parse(row);
            const value = jsonObject.data[0][col];
            console.log(`\t ${col} - ${JSON.stringify(value)}`);
        }
        rowCount++;
    }
}

main().catch((err) => {
    console.error(err);
});

Use the ingestFromFile() API to ingest the stormevents.csv file. Place the stormevents.csv file in the same location as your script. Since our input is a CSV file, use format: DataFormat.CSV in the ingestion properties.

Add and ingestion section using the following lines to the end of main().

const ingestProperties = new IngestionProperties({
    database: databaseName,
    table: tableName,
    format: DataFormat.CSV
});
//Ingest section
console.log("Ingesting data from a file");
await ingestClient.ingestFromFile(".\\stormevents.csv", ingestProperties);
ingestClient.close();

Let’s also query the new number of rows and the most recent row after the ingestion. Add the following lines after the ingestion command:

console.log(`New number of rows in ${tableName}`);
result = await kustoClient.executeQuery(databaseName, `${tableName} | count`);
printResultAsValueList(result);
console.log(`Example line from ${tableName}`);
result = await kustoClient.executeQuery(databaseName, `${tableName} | top 1 by EndTime`);
printResultAsValueList(result);

The code sample includes a service method printResultsAsValueList() for printing query results.

package com.example;

import java.io.FileWriter;

import com.azure.identity.DefaultAzureCredential;
import com.azure.identity.DefaultAzureCredentialBuilder;
import com.microsoft.azure.kusto.data.Client;
import com.microsoft.azure.kusto.data.ClientFactory;
import com.microsoft.azure.kusto.data.KustoOperationResult;
import com.microsoft.azure.kusto.data.KustoResultSetTable;
import com.microsoft.azure.kusto.data.KustoResultColumn;
import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
import com.microsoft.azure.kusto.ingest.IngestClientFactory;
import com.microsoft.azure.kusto.ingest.IngestionProperties;
import com.microsoft.azure.kusto.ingest.ManagedStreamingIngestClient;
import com.microsoft.azure.kusto.ingest.IngestionProperties.DataFormat;
import com.microsoft.azure.kusto.ingest.source.FileSourceInfo;

public class BatchIngestion {
    public static void main(String[] args) throws Exception {

        String clusterUri = "<KustoClusterQueryURI>";
        String ingestionUri = "<KustoClusterQueryIngestURI>";
        String databaseName = "<databaseName>";
        String table = "MyStormEvents";

        ConnectionStringBuilder clusterKcsb = ConnectionStringBuilder.createWithUserPrompt(clusterUri);
        ConnectionStringBuilder ingestionKcsb = ConnectionStringBuilder.createWithUserPrompt(ingestionUri);

        try (
                Client kustoClient = ClientFactory.createClient(clusterKcsb)) {

            String query = table + " | count";
            KustoOperationResult results = kustoClient.execute(databaseName, query);
            KustoResultSetTable primaryResults = results.getPrimaryResults();
            System.out.println("\nNumber of rows in " + table + " BEFORE ingestion:");
            printResultsAsValueList(primaryResults);
        }
    }

    public static void printResultsAsValueList(KustoResultSetTable results) {
        while (results.next()) {
            KustoResultColumn[] columns = results.getColumns();
            for (int i = 0; i < columns.length; i++) {
                System.out.println("\t" + columns[i].getColumnName() + " - "
                        + (results.getObject(i) == null ? "None" : results.getString(i)));
            }
        }
    }
}

Use the ingestFromFile() method to ingest the stormevents.csv file. Place the stormevents.csv file in the same location as your script. Since our input is a CSV file, use ingestionProperties.setDataFormat(DataFormat.CSV) in the ingestion properties.

Add and ingestion section using the following lines to the end of main().

// Ingestion section
try (
        ManagedStreamingIngestClient ingestClient =  IngestClientFactory
                .createManagedStreamingIngestClient(clusterKcsb, ingestionKcsb)) {
    System.out.println("Ingesting data from a file");
    String filePath = "stormevents.csv";
    IngestionProperties ingestionProperties = new IngestionProperties(databaseName, table);
    ingestionProperties.setDataFormat(DataFormat.CSV);
    FileSourceInfo fileSourceInfo = new FileSourceInfo(filePath, 0);
    ingestClient.ingestFromFile(fileSourceInfo, ingestionProperties);
} catch (Exception e) {
    // TODO: handle exception
    System.out.println("Error: " + e);
}

Let’s also query the new number of rows and the most recent row after the ingestion. Add the following lines after the ingestion command:

query = table + " | count";
results = kustoClient.execute(databaseName, query);
primaryResults = results.getPrimaryResults();
System.out.println("\nNumber of rows in " + table + " AFTER ingestion:");
printResultsAsValueList(primaryResults);

query = table + " | top 1 by EndTime";
results = kustoClient.execute(databaseName, query);
primaryResults = results.getPrimaryResults();
System.out.println("\nExample line from " + table);
printResultsAsValueList(primaryResults);

To ingest the stream from memory, call the IngestFromStreamAsync() method.

Replace the ingestion section with the following code:

// Ingestion section
Console.WriteLine("Ingesting data from memory");
var singleLine = "2018-01-26 00:00:00.0000000,2018-01-27 14:00:00.0000000,MEXICO,0,0,Unknown,'{}'";
byte[] byteArray = Encoding.UTF8.GetBytes(singleLine);
using (MemoryStream stream = new MemoryStream(byteArray))
   {
    var streamSourceOptions = new StreamSourceOptions
    {
        LeaveOpen = false
    };
    ingestClient.IngestFromStreamAsync(stream, ingestProperties, streamSourceOptions).Wait();
   }

To ingest the stream from memory, call the ingest_from_stream() API.

Replace the ingestion section with the following code:

# Ingestion section
print("Ingesting data from memory")
single_line = '2018-01-26 00:00:00.0000000,2018-01-27 14:00:00.0000000,MEXICO,0,0,Unknown,"{}"'
string_stream = io.StringIO(single_line)
ingest_properties = IngestionProperties(database_name, table_name, DataFormat.CSV)
# when possible provide the size of the raw data
stream_descriptor = StreamDescriptor(string_stream, is_compressed=False, size=len(single_line))
ingest_client.ingest_from_stream(stream_descriptor, ingest_properties)

To ingest the stream from memory, call the ingestFromStream() API.

Replace the ingestion section with the following code:

//Ingest section
console.log('Ingesting data from memory');
const singleLine = '2018-01-26 00:00:00.0000000,2018-01-27 14:00:00.0000000,MEXICO,0,0,Unknown,"{}"'
await ingestClient.ingestFromStream(Buffer.from(singleLine), ingestProperties)
ingestClient.close();

To ingest the stream from memory, call the ingestFromStream() API.

Replace the ingestion section with the following code:

String singleLine = "2018-01-26 00:00:00.0000000,2018-01-27 14:00:00.0000000,MEXICO,0,0,Unknown,\"{}\"";
ByteArrayInputStream inputStream = new ByteArrayInputStream(singleLine.getBytes(StandardCharsets.UTF_8));
try (
        ManagedStreamingIngestClient ingestClient = (ManagedStreamingIngestClient) IngestClientFactory
                .createManagedStreamingIngestClient(clusterKcsb, ingestionKcsb)) {
    System.out.println("Ingesting data from a byte array");
    IngestionProperties ingestionProperties = new IngestionProperties(databaseName, table);
    ingestionProperties.setDataFormat(DataFormat.CSV);
    StreamSourceInfo streamSourceInfo = new StreamSourceInfo(inputStream);
    ingestClient.ingestFromStream(streamSourceInfo, ingestionProperties);
} catch (Exception e) {
    // TODO: handle exception
    System.out.println("Error: " + e);
}