Konfigurera datainmatning för streaming i ditt Azure Data Explorer-kluster

Direktuppspelningsinmatning är användbart för att läsa in data när du behöver korta svarstider mellan inmatning och fråga. Överväg att använda strömningsinmatning i följande scenarier:

• Svarstid på mindre än en sekund krävs.
• Optimera driftbearbetningen av många tabeller där dataströmmen till varje tabell är relativt liten (några poster per sekund), men den totala datainmatningsvolymen är hög (tusentals poster per sekund).

Om dataströmmen i varje tabell är hög (över 4 GB per timme) bör du överväga att använda batchinmatning.

Mer information om olika inmatningsmetoder finns i översikten över datainmatning.

Välj lämplig typ av direktuppspelningsinmatning

Två typer av direktuppspelningsinmatning stöds:

Inmatningstyp	Description
Dataanslutning	Event Hub-, IoT Hub- och Event Grid-dataanslutningar kan använda strömmande inmatning, förutsatt att den är aktiverad på klusternivå. Beslutet att använda strömningsinmatning görs enligt den princip för strömningsinmatning som konfigurerats i måltabellen. Information om hur du hanterar dataanslutningar finns i Event Hub, IoT Hub och Event Grid.
Anpassad inmatning	Anpassad inmatning kräver att du skriver ett program som använder ett av Azure Data Explorer klientbibliotek. Använd informationen i det här avsnittet för att konfigurera anpassad inmatning. Du kanske också tycker att C#-exempelprogrammet för direktuppspelning är användbart.

Använd följande tabell för att hjälpa dig att välja den inmatningstyp som är lämplig för din miljö:

Kriterium	Dataanslutning	Anpassad inmatning
Datafördröjning mellan inmatningsinitiering och tillgängliga data för fråga	Längre fördröjning	Kortare fördröjning
Utvecklingskostnader	Snabb och enkel installation, inga utvecklingskostnader	Höga utvecklingskostnader för att skapa ett program som matar in data, hanterar fel och säkerställer datakonsekvens

Anteckning

Du kan hantera processen för att aktivera och inaktivera strömmande inmatning i klustret med hjälp av Azure Portal eller programmatiskt i C#. Om du använder C# för ditt anpassade program kan det vara enklare att använda den programmatiska metoden.

Förutsättningar

• En Azure-prenumeration. Skapa ett kostnadsfritt Azure-konto.

Prestanda- och driftsöverväganden

De främsta bidragsgivarna som kan påverka strömmande inmatning är:

• VM- och klusterstorlek: Strömmande inmatningsprestanda och kapacitetsskalor med ökade storlekar på virtuella datorer och kluster. Antalet samtidiga inmatningsbegäranden är begränsat till sex per kärna. För SKU:er med 16 kärnor, till exempel D14 och L16, är den maximala belastningen som stöds 96 samtidiga inmatningsbegäranden. För två kärn-SKU:er, till exempel D11, är den maximala belastningen som stöds 12 samtidiga inmatningsbegäranden.
• Datastorleksgräns: Datastorleksgränsen för en begäran om strömmande inmatning är 4 MB. Detta inkluderar alla data som skapats för uppdateringsprinciper under inmatningen.
• Schemauppdateringar: Schemauppdateringar, till exempel skapande och ändring av tabeller och inmatningsmappningar, kan ta upp till fem minuter för tjänsten för direktuppspelning. Mer information finns i Strömmande inmatning och schemaändringar.
• SSD-kapacitet: Aktivera strömmande inmatning i ett kluster, även om data inte matas in via strömning, använder en del av den lokala SSD-disken på klusterdatorerna för strömmande inmatningsdata och minskar lagringen som är tillgänglig för frekvent cache.

Aktivera strömningsinmatning i klustret

Innan du kan använda direktuppspelningsinmatning måste du aktivera funktionen i klustret och definiera en princip för strömningsinmatning. Du kan aktivera funktionen när du skapar klustret eller lägga till det i ett befintligt kluster.

Varning

Granska begränsningarna innan du aktiverar strömningsinmatning.

Aktivera direktuppspelningsinmatning när du skapar ett nytt kluster

Du kan aktivera direktuppspelningsinmatning när du skapar ett nytt kluster med hjälp av Azure Portal eller programmatiskt i C#.

Portal

När du skapar ett kluster med hjälp av stegen i Skapa ett Azure Data Explorer-kluster och -databas går du till fliken Konfigurationer och väljer Direktuppspelningsinmatning>på.

C#

Om du vill aktivera direktuppspelningsinmatning när du skapar ett nytt Azure Data Explorer-kluster kör du följande kod:

using System.Threading.Tasks;
using Microsoft.Azure.Management.Kusto; // Required package Microsoft.Azure.Management.Kusto
using Microsoft.Azure.Management.Kusto.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory; // Required package Microsoft.IdentityModel.Clients.ActiveDirectory
using Microsoft.Rest;

namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string clusterName = "<clusterName>";
            string resourceGroupName = "<resourceGroupName>";
            string subscriptionId = "<subscriptionId>";
            string location = "<location>";
            string skuName = "<skuName>";
            string tier = "<tier>";

            var authenticationContext = new AuthenticationContext($"https://login.windows.net/{appTenant}");
            var credential = new ClientCredential(appId, appKey);
            var result = await authenticationContext.AcquireTokenAsync(resource: "https://management.core.windows.net/", clientCredential: credential);

            var credentials = new TokenCredentials(result.AccessToken, result.AccessTokenType);
            var kustoManagementClient = new KustoManagementClient(credentials)
            {
                SubscriptionId = subscriptionId
            };

            var cluster = new Cluster(location, new AzureSku(skuName, tier), enableStreamingIngest:true);
            await kustoManagementClient.Clusters.CreateOrUpdateAsync(resourceGroupName, clusterName, cluster);
        }
    }
}

Aktivera strömningsinmatning i ett befintligt kluster

Om du har ett befintligt kluster kan du aktivera direktuppspelning med hjälp av Azure Portal eller programmatiskt i C#.

Portal

1. I Azure Portal går du till ditt Azure Data Explorer-kluster.

2. I Inställningar väljer du Konfigurationer.

3. I fönstret Konfigurationer väljer du På för att aktivera strömningsinmatning.

4. Välj Spara.

   

C#

Du kan aktivera direktuppspelningsinmatning när du skapar ett nytt Azure-Data Explorer kluster.

using System.Threading.Tasks;
using Microsoft.Azure.Management.Kusto; // Required package Microsoft.Azure.Management.Kusto
using Microsoft.Azure.Management.Kusto.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory; // Required package Microsoft.IdentityModel.Clients.ActiveDirectory
using Microsoft.Rest;

namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string clusterName = "<clusterName>";
            string resourceGroupName = "<resourceGroupName>";
            string subscriptionId = "<subscriptionId>";

            var authenticationContext = new AuthenticationContext($"https://login.windows.net/{appTenant}");
            var credential = new ClientCredential(appId, appKey);
            var result = await authenticationContext.AcquireTokenAsync(resource: "https://management.core.windows.net/", clientCredential: credential);

            var credentials = new TokenCredentials(result.AccessToken, result.AccessTokenType);
            var kustoManagementClient = new KustoManagementClient(credentials)
            {
                SubscriptionId = subscriptionId
            };

            var clusterUpdateParameters = new ClusterUpdate(enableStreamingIngest: true);
            await kustoManagementClient.Clusters.UpdateAsync(resourceGroupName, clusterName, clusterUpdateParameters);
        }
    }
}

Skapa en måltabell och definiera principen

Skapa en tabell för att ta emot strömmande inmatningsdata och definiera dess relaterade princip med hjälp av Azure Portal eller programmatiskt i C#.

Portal

1. I Azure Portal navigerar du till klustret.

2. Välj Fråga.

   

3. Om du vill skapa tabellen som ska ta emot data via strömmande inmatning kopierar du följande kommando till frågefönstret och väljer Kör.

   .create table TestTable (TimeStamp: datetime, Name: string, Metric: int, Source:string)
   

   

4. Kopiera något av följande kommandon till frågefönstret och välj Kör. Detta definierar principen för strömmande inmatning i den tabell som du skapade eller i databasen som innehåller tabellen.

   Tips

   En princip som definieras på databasnivå gäller för alla befintliga och framtida tabeller i databasen. När du aktiverar principen på databasnivå behöver du inte aktivera den per tabell.

   • Om du vill definiera principen i tabellen som du skapade använder du:

     .alter table TestTable policy streamingingestion enable
     

   • Om du vill definiera principen för databasen som innehåller den tabell som du skapade använder du:

     .alter database StreamingTestDb policy streamingingestion enable
     

   

C#

using System.Threading.Tasks;
using Kusto.Data; // Requires Package Microsoft.Azure.Kusto.Data
using Kusto.Data.Common;
using Kusto.Data.Net.Client;

namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        // Create Kusto connection string with App Authentication
        var connectionStringBuilder = new KustoConnectionStringBuilder(clusterPath)
            .WithAadApplicationKeyAuthentication(
                applicationClientId: appId,
                applicationKey: appKey,
                authority: appTenant
            );
        using var client = KustoClientFactory.CreateCslAdminProvider(connectionStringBuilder);
        
        var tableName = "<tableName>";
        var tableSchema = new TableSchema(
            tableName,
            new ColumnSchema[]
            {
                new("TimeStamp", "System.DateTime"),
                new("Name", "System.String"),
                new("Metric", "System.int"),
                new("Source", "System.String"),
            });
        var tableCreateCommand = CslCommandGenerator.GenerateTableCreateCommand(tableSchema);
        var tablePolicyAlterCommand = CslCommandGenerator.GenerateTableAlterStreamingIngestionPolicyCommand(tableName, isEnabled: true);
        
        await client.ExecuteControlCommandAsync(tableCreateCommand);
        await client.ExecuteControlCommandAsync(tablePolicyAlterCommand);
    }
}

Skapa ett strömmande inmatningsprogram för att mata in data till klustret

Skapa ditt program för att mata in data till klustret med det språk du föredrar.

C#

using System.IO;
using System.Threading.Tasks;
using Kusto.Data; // Requires Package Microsoft.Azure.Kusto.Data
using Kusto.Data.Common;
using Kusto.Ingest; // Requires Package Microsoft.Azure.Kusto.Ingest

namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        // Create Kusto connection string with App Authentication
        var connectionStringBuilder = new KustoConnectionStringBuilder(clusterPath)
            .WithAadApplicationKeyAuthentication(
                applicationClientId: appId,
                applicationKey: appKey,
                authority: appTenant
            );
        // Create a disposable client that will execute the ingestion
        using var client = KustoIngestFactory.CreateStreamingIngestClient(connectionStringBuilder);
        // Ingest from a compressed file
        var fileStream = File.Open("MyFile.gz", FileMode.Open);
        // Initialize client properties
        var ingestionProperties = new KustoIngestionProperties(databaseName: "<databaseName>", tableName: "<tableName>");
        // Create source options
        var sourceOptions = new StreamSourceOptions { CompressionType = DataSourceCompressionType.GZip, };
        // Ingest from stream
        await client.IngestFromStreamAsync(fileStream, ingestionProperties, sourceOptions);
    }
}

Python

from azure.kusto.data import KustoConnectionStringBuilder, DataFormat
from azure.kusto.ingest import IngestionProperties, KustoStreamingIngestClient

clusterPath = "https://<clusterName>.<region>.kusto.windows.net"
appId = "<appId>"
appKey = "<appKey>"
appTenant = "<appTenant>"
dbName = "<dbName>"
tableName = "<tableName>"

csb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
    clusterPath,
    appId,
    appKey,
    appTenant
)
client = KustoStreamingIngestClient(csb)

ingestionProperties = IngestionProperties(
    database=dbName,
    table=tableName,
    data_format=DataFormat.CSV
)

# Ingest from file
# Automatically detects gz format
client.ingest_from_file("MyFile.gz", ingestion_properties=ingestionProperties) 

Node.js

// Load modules using ES6 import statements:
import { DataFormat, IngestionProperties, StreamingIngestClient } from "azure-kusto-ingest";
import { KustoConnectionStringBuilder } from "azure-kusto-data";

// For earlier version, load modules using require statements:
// const IngestionProperties = require("azure-kusto-ingest").IngestionProperties;
// const KustoConnectionStringBuilder = require("azure-kusto-data").KustoConnectionStringBuilder;
// const {DataFormat} = require("azure-kusto-ingest").IngestionPropertiesEnums;
// const StreamingIngestClient = require("azure-kusto-ingest").StreamingIngestClient;

const clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
const appId = "<appId>";
const appKey = "<appKey>";
const appTenant = "<appTenant>";
const dbName = "<dbName>";
const tableName = "<tableName>";
const mappingName = "<mappingName>"; // Required for JSON formatted files

const ingestionProperties = new IngestionProperties({
    database: dbName, // Your database
    table: tableName, // Your table
    format: DataFormat.JSON,
    ingestionMappingReference: mappingName
});

// Initialize client with engine endpoint
const client = new StreamingIngestClient(
    KustoConnectionStringBuilder.withAadApplicationKeyAuthentication(
        clusterPath,
        appId,
        appKey,
        appTenant
    ),
    ingestionProperties
);

// Automatically detects gz format
await client.ingestFromFile("MyFile.gz", ingestionProperties);

Kör

import (
    "context"
    "github.com/Azure/azure-kusto-go/kusto"
    "github.com/Azure/azure-kusto-go//azure/data-explorer/kusto/ingest"
    "github.com/Azure/go-autorest/autorest/azure/auth"
)

func ingest() {
    clusterPath := "https://<clusterName>.kusto.windows.net"
    appId := "<appId>"
    appKey := "<appKey>"
    appTenant := "<appTenant>"
    dbName := "<dbName>"
    tableName := "<tableName>"
    mappingName := "<mappingName>" // Optional, can be nil

    // Creates a Kusto Authorizer using your client identity, secret, and tenant identity.
    // You may also uses other forms of authorization, see GoDoc > Authorization type.
    // auth package is: "github.com/Azure/go-autorest/autorest/azure/auth"
    authorizer := kusto.Authorization{
        Config: auth.NewClientCredentialsConfig(appId, appKey, appTenant),
    }

    // Create a client
    client, err := kusto.New(clusterPath, authorizer)
    if err != nil {
        panic("add error handling")
    }

    // Create an ingestion instance
    // Pass the client, the name of the database, and the name of table you wish to ingest into.
    in, err := ingest.New(client, dbName, tableName)
    if err != nil {
        panic("add error handling")
    }

    // Go currently only supports streaming from a byte array with a maximum size of 4 MB.
    jsonEncodedData := []byte("{\"a\":  1, \"b\":  10}\n{\"a\":  2, \"b\":  20}")

    // Ingestion from a stream commits blocks of fully formed data encodes (JSON, AVRO, ...) into Kusto:
    if err := in.Stream(context.Background(), jsonEncodedData, ingest.JSON, mappingName); err != nil {
        panic("add error handling")
    }
}

Java

import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
import com.microsoft.azure.kusto.ingest.IngestClient;
import com.microsoft.azure.kusto.ingest.IngestClientFactory;
import com.microsoft.azure.kusto.ingest.IngestionProperties;
import com.microsoft.azure.kusto.ingest.result.OperationStatus;
import com.microsoft.azure.kusto.ingest.source.CompressionType;
import com.microsoft.azure.kusto.ingest.source.StreamSourceInfo;
import java.io.FileInputStream;
import java.io.InputStream;

public class FileIngestion {
    public static void main(String[] args) throws Exception {
        String clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
        String appId = "<appId>";
        String appKey = "<appKey>";
        String appTenant = "<appTenant>";
        String dbName = "<dbName>";
        String tableName = "<tableName>";

        // Build connection string and initialize
        ConnectionStringBuilder csb =
            ConnectionStringBuilder.createWithAadApplicationCredentials(
                clusterPath,
                appId,
                appKey,
                appTenant
            );

        // Initialize client and its properties
        IngestClient client = IngestClientFactory.createClient(csb);
        IngestionProperties ingestionProperties =
            new IngestionProperties(
                dbName,
                tableName
            );

        // Ingest from a compressed file
        // Create Source info
        InputStream zipInputStream = new FileInputStream("MyFile.gz");
        StreamSourceInfo zipStreamSourceInfo = new StreamSourceInfo(zipInputStream);
        // If the data is compressed
        zipStreamSourceInfo.setCompressionType(CompressionType.gz);
        // Ingest from stream
        OperationStatus status = client.ingestFromStream(zipStreamSourceInfo, ingestionProperties).getIngestionStatusCollection().get(0).status;
    }
}

Inaktivera direktuppspelningsinmatning i klustret

Varning

Det kan ta några timmar att inaktivera strömmande inmatning.

Innan du inaktiverar direktuppspelningsinmatning i azure-Data Explorer-klustret släpper du principen för strömningsinmatning från alla relevanta tabeller och databaser. Borttagningen av principen för direktuppspelning utlöser omorganisering av data i Azure Data Explorer-klustret. Strömmande inmatningsdata flyttas från den ursprungliga lagringen till permanent lagring i kolumnlagret (omfattningar eller shards). Den här processen kan ta mellan några sekunder och några timmar, beroende på mängden data i den ursprungliga lagringen.

Ta bort principen för strömningsinmatning

Du kan ta bort principen för strömningsinmatning med hjälp av Azure Portal eller programmatiskt i C#.

Portal

1. I Azure Portal går du till ditt Azure Data Explorer-kluster och väljer Fråga.

2. Om du vill ta bort den strömmande inmatningsprincipen från tabellen kopierar du följande kommando till frågefönstret och väljer Kör.

   .delete table TestTable policy streamingingestion
   

   

3. I Inställningar väljer du Konfigurationer.

4. I fönstret Konfigurationer väljer du Av för att inaktivera strömningsinmatning.

5. Välj Spara.

   

C#

Om du vill ta bort principen för strömningsinmatning från tabellen kör du följande kod:

using System.Threading.Tasks;
using Kusto.Data; // Requires Package Microsoft.Azure.Kusto.Data
using Kusto.Data.Common;
using Kusto.Data.Net.Client;

namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        // Create Kusto connection string with App Authentication
        var connectionStringBuilder = new KustoConnectionStringBuilder(clusterPath)
            .WithAadApplicationKeyAuthentication(
                applicationClientId: appId,
                applicationKey: appKey,
                authority: appTenant
            );
        using var client = KustoClientFactory.CreateCslAdminProvider(connectionStringBuilder);
        var tablePolicyDropCommand = CslCommandGenerator.GenerateTableStreamingIngestionPolicyDropCommand("<dbName>", "<tableName>");
        await client.ExecuteControlCommandAsync(tablePolicyDropCommand);
    }
}

Om du vill inaktivera strömningsinmatning i klustret kör du följande kod:

using System.Threading.Tasks;
using Microsoft.Azure.Management.Kusto; // Required package Microsoft.Azure.Management.Kusto
using Microsoft.Azure.Management.Kusto.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory; // Required package Microsoft.IdentityModel.Clients.ActiveDirectory
using Microsoft.Rest;

namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string clusterName = "<clusterName>";
            string resourceGroupName = "<resourceGroupName>";
            string subscriptionId = "<subscriptionId>";
            
            var authenticationContext = new AuthenticationContext($"https://login.windows.net/{appTenant}");
            var credential = new ClientCredential(appId, appKey);
            var result = await authenticationContext.AcquireTokenAsync(resource: "https://management.core.windows.net/", clientCredential: credential);
            
            var credentials = new TokenCredentials(result.AccessToken, result.AccessTokenType);
            var kustoManagementClient = new KustoManagementClient(credentials)
            {
                SubscriptionId = subscriptionId
        };
        
            var clusterUpdateParameters = new ClusterUpdate(enableStreamingIngest: false);
            await kustoManagementClient.Clusters.UpdateAsync(resourceGroupName, clusterName, clusterUpdateParameters);
        }
    }
}

Begränsningar

• Datamappningar måste skapas i förväg för användning vid strömningsinmatning. Enskilda begäranden om direktuppspelningsinmatning har inte plats för infogade datamappningar.
• Det går inte att ange utrymmestaggar för strömmande inmatningsdata.
• Uppdatera princip. Uppdateringsprincipen kan endast referera till nyligen inmatade data i källtabellen och inte andra data eller tabeller i databasen.
• Om strömningsinmatning är aktiverat på ett kluster som används som ledare för efterföljande databaser, måste strömmande inmatning aktiveras på följande kluster samt för att följa strömmande inmatningsdata. Samma sak gäller om klusterdata delas via Data Share.

Nästa steg

• Fråga efter data i Azure Data Explorer