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 by using the following commands:

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

using System.Data;

using Kusto.Data;
using Kusto.Data.Common;
using Kusto.Data.Net.Client;
using Kusto.Ingest;
using Kusto.Ingest.Common;


using Azure.Identity;

namespace BatchIngest;

class BatchIngest
{
    static async Task Main()
    {
       var tokenCredential = new InteractiveBrowserCredential();
       var clusterUri = "<your cluster>"; // e.g., "https://<your_cluster_name>.<region>.kusto.chinacloudapi.cn"
       var clusterKcsb = new KustoConnectionStringBuilder(clusterUri).WithAadAzureTokenCredentialsAuthentication(tokenCredential);
        using var kustoClient = KustoClientFactory.CreateCslQueryProvider(clusterKcsb);
        var database = "<your database>";
        var table = "MyStormEvents";
        var query = table + " | count";
        using (var response = await kustoClient.ExecuteQueryAsync(database, query, null))
        {
            Console.WriteLine("\nNumber of rows in " + table + " BEFORE ingestion:");
            PrintResultsAsValueList(response);
        }
    }

    static void PrintResultsAsValueList(IDataReader response)
    {
        while (response.Read())
        {
            for (var i = 0; i < response.FieldCount; i++)
            {
                object val = response.GetValue(i);
                string value = val.ToString() ?? "None";
                Console.WriteLine("\t{0} - {1}", response.GetName(i), value);
            }
        }
    }
}

Stream a file for ingestion

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().

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

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 = await kustoClient.ExecuteQueryAsync(databaseName, tableName + " | count", new ClientRequestProperties());
PrintResultAsValueList(result);

Console.WriteLine("Example line from " + tableName);
result = await kustoClient.ExecuteQueryAsync(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()

Stream a file for ingestion

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);
});

Stream a file for ingestion

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)));
            }
        }
    }
}

Stream a file for ingestion

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);

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

Add the Kusto libraries by using the following commands:

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

using System.Data;

using Kusto.Data;
using Kusto.Data.Common;
using Kusto.Data.Net.Client;
using Kusto.Ingest.V2;
using Kusto.Ingest.Common;

using Azure.Identity;

namespace BatchIngest;

class BatchIngest
{
    static async Task Main()
    {
       var tokenCredential = new InteractiveBrowserCredential();
       var clusterUri = "<your cluster>"; // e.g., "https://<your_cluster_name>.<region>.kusto.chinacloudapi.cn"
       var clusterKcsb = new KustoConnectionStringBuilder(clusterUri).WithAadAzureTokenCredentialsAuthentication(tokenCredential);
        var database = "<your database>";
        var table = "MyStormEvents";
        var query = table + " | count";
        using var kustoClient = KustoClientFactory.CreateCslQueryProvider(clusterKcsb);
        using (var response = await kustoClient.ExecuteQueryAsync(database, query, null))
        {
            Console.WriteLine("\nNumber of rows in " + table + " BEFORE ingestion:");
            PrintResultsAsValueList(response);
        }
    }

    static void PrintResultsAsValueList(IDataReader response)
    {
        while (response.Read())
        {
            for (var i = 0; i < response.FieldCount; i++)
            {
                object val = response.GetValue(i);
                string value = val.ToString() ?? "None";
                Console.WriteLine("\t{0} - {1}", response.GetName(i), value);
            }
        }
    }
}

Stream a file for ingestion

Use the IngestAsync 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 DataSourceFormat.csv as the format in the FileSource.

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

using var ingestClient = ManagedStreamingIngestClientBuilder.Create(new Uri(clusterUri)).WithAuthentication(tokenCredential).Build();
var fileSource = new FileSource(.\\stormevents.csv, DataSourceFormat.csv);
await ingestClient.IngestAsync(fileSource, database, table);

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 = await kustoClient.ExecuteQueryAsync(databaseName, tableName + " | count", new ClientRequestProperties());
PrintResultAsValueList(result);

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

Not applicable.

Not applicable.

Not applicable.

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 var ingestClient = KustoIngestFactory.CreateManagedStreamingIngestClient(clusterKcsb);
    using var stream = new MemoryStream(byteArray);

    var streamSourceOptions = new StreamSourceOptions
    {
        LeaveOpen = false
    };

    await ingestClient.IngestFromStreamAsync(stream, ingestProperties, streamSourceOptions);

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);
}

    // 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 var ingestClient = QueuedIngestClientBuilder.Create(new Uri(clusterUri)).WithAuthentication(tokenCredential).Build();
    using var ingestClient = KustoIngestFactory.CreateManagedStreamingIngestClient(clusterKcsb);
    using var stream = new MemoryStream(byteArray);
    var streamSource = new StreamSource(.\\stormevents.csv, DataSourceCompressionType.None, DataSourceFormat.csv);
    await ingestClient.IngestAsync(streamSource, database, table);

Not applicable.

Not applicable.

Not applicable.