Konfigurace příjmu streamovaných dat v clusteru Azure Data Exploreru

Streamování příjmu dat je užitečné pro načítání dat, když potřebujete nízkou latenci mezi příjmem dat a dotazem. Použití příjmu dat streamování zvažte v následujících scénářích:

• Vyžaduje se latence menší než sekunda.
• Optimalizace provozního zpracování mnoha tabulek, kde je stream dat do každé tabulky relativně malý (několik záznamů za sekundu), ale celkový objem příjmu dat je vysoký (tisíce záznamů za sekundu).

Pokud je datový proud dat do každé tabulky vysoký (více než 4 GB za hodinu), zvažte použití dávkového příjmu dat.

Další informace o různých metodách příjmu dat najdete v tématu Přehled příjmu dat.

Zvolte vhodný typ příjmu dat streamování.

Podporují se dva typy příjmu dat streamování:

Typ příjmu dat	Popis
Datové připojení	Datová připojení centra událostí, IoT Hub a Event Gridu můžou využívat příjem dat streamování za předpokladu, že je povolený na úrovni clusteru. Rozhodnutí o použití příjmu dat streamování se provádí podle zásad příjmu dat streamování nakonfigurovaných v cílové tabulce. Informace o správě datových připojení najdete v tématu Centrum událostí, IoT Hub a Event Grid.
Vlastní příjem dat	Vlastní příjem dat vyžaduje, abyste napsali aplikaci, která používá jednu z klientských knihoven Azure Data Explorer. Informace v tomto tématu slouží ke konfiguraci vlastního příjmu dat. Může se vám také hodit ukázková aplikace pro příjem dat streamování v jazyce C# .

Následující tabulka vám pomůže vybrat typ příjmu dat, který je vhodný pro vaše prostředí:

Kritérium	Datové připojení	Vlastní příjem dat
Zpoždění dat mezi zahájením příjmu dat a daty dostupnými pro dotaz	Delší zpoždění	Kratší zpoždění
Režie na vývoj	Rychlé a snadné nastavení, bez režie na vývoj	Vysoké režijní náklady na vývoj při vytváření ingestování dat aplikace, zpracování chyb a zajištění konzistence dat

Poznámka

Proces povolení a zakázání příjmu streamovaných dat v clusteru můžete spravovat pomocí Azure Portal nebo programově v jazyce C#. Pokud pro vlastní aplikaci používáte jazyk C#, může být vhodnější použít programový přístup.

Požadavky

• Předplatné Azure. Vytvořte si bezplatný účet Azure.

Důležité informace o výkonu a provozu

Mezi hlavní přispěvatele, kteří můžou mít vliv na příjem dat streamování, patří:

• Velikost virtuálního počítače a clusteru: Výkon a kapacita streamování se škálují se zvýšenými velikostmi virtuálních počítačů a clusterů. Počet souběžných žádostí o příjem dat je omezený na šest na jádro. Například u skladových položek s 16 jádry, jako jsou D14 a L16, je maximální podporované zatížení 96 souběžných požadavků na příjem dat. U dvou základních skladových položek, jako je D11, je maximální podporovaná zátěž 12 souběžných požadavků na příjem dat.
• Limit velikosti dat: Limit velikosti dat pro požadavek na ingestování streamování je 4 MB. To zahrnuje všechna data vytvořená pro zásady aktualizace během příjmu dat.
• Aktualizace schématu: Aktualizace schématu, jako je vytváření a úpravy tabulek a mapování příjmu dat, můžou u služby příjmu dat streamování trvat až pět minut. Další informace najdete v tématu Příjem dat streamování a změny schématu.
• Kapacita SSD: Povolení příjmu streamovaných dat v clusteru i v případě, že data nejsou ingestovaná prostřednictvím streamování, využívá část místního disku SSD počítačů clusteru pro streamování dat a snižuje velikost dostupného úložiště pro horkou mezipaměť.

Povolení příjmu streamovaných dat v clusteru

Než budete moct použít příjem dat streamování, musíte tuto funkci v clusteru povolit a definovat zásady příjmu dat streamování. Tuto funkci můžete povolit při vytváření clusteru nebo ho přidat do existujícího clusteru.

Upozornění

Před povolením příjmu dat streamování si projděte omezení .

Povolení příjmu dat streamování při vytváření nového clusteru

Příjem dat streamování můžete povolit při vytváření nového clusteru pomocí Azure Portal nebo programově v jazyce C#.

Azure Portal

Při vytváření clusteru pomocí kroků v tématu Vytvoření clusteru a databáze Azure Data Explorer na kartě Konfigurace vyberte Příjem dat streamování>Zapnuto.

C#

Pokud chcete povolit příjem dat streamování při vytváření nového clusteru Azure Data Explorer, spusťte následující kód:

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

Povolení příjmu streamovaných dat v existujícím clusteru

Pokud máte existující cluster, můžete povolit příjem dat streamování pomocí Azure Portal nebo programově v jazyce C#.

Azure Portal

1. V Azure Portal přejděte do clusteru Azure Data Explorer.

2. V Nastavení vyberte Konfigurace.

3. V podokně Konfigurace vyberte Zapnuto a povolte příjem dat streamování.

4. Vyberte Uložit.

   

C#

Příjem dat streamování můžete povolit při vytváření nového clusteru Azure Data Explorer.

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

Vytvoření cílové tabulky a definování zásad

Vytvořte tabulku pro příjem streamovaných dat příjmu dat a definujte související zásady pomocí Azure Portal nebo programově v jazyce C#.

Azure Portal

1. V Azure Portal přejděte do clusteru.

2. Vyberte Dotaz.

   

3. Pokud chcete vytvořit tabulku, která bude přijímat data prostřednictvím příjmu dat streamování, zkopírujte následující příkaz do podokna Dotaz a vyberte Spustit.

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

   

4. Zkopírujte jeden z následujících příkazů do podokna Dotaz a vyberte Spustit. Tím se definují zásady příjmu dat streamování v tabulce, kterou jste vytvořili, nebo v databázi, která tabulku obsahuje.

   Tip

   Zásada definovaná na úrovni databáze se vztahuje na všechny existující a budoucí tabulky v databázi. Když povolíte zásadu na úrovni databáze, není nutné ji povolovat pro jednotlivé tabulky.

   • Pokud chcete definovat zásady pro tabulku, kterou jste vytvořili, použijte:

     .alter table TestTable policy streamingingestion enable
     

   • Pokud chcete definovat zásady pro databázi obsahující tabulku, kterou jste vytvořili, použijte:

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

Vytvoření aplikace pro příjem streamovaných dat do clusteru

Vytvořte aplikaci pro příjem dat do clusteru pomocí preferovaného jazyka.

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

Přejít

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

Zákaz příjmu dat streamování v clusteru

Upozornění

Zakázání příjmu dat streamování může trvat několik hodin.

Před zakázáním příjmu dat streamování v clusteru Azure Data Explorer odstraňte zásady příjmu dat ze všech relevantních tabulek a databází. Odebráním zásad příjmu dat streamování se aktivuje změna uspořádání dat v clusteru Azure Data Explorer. Streamovaná data příjmu dat se přesunou z počátečního úložiště do trvalého úložiště ve sloupcovém úložišti (rozsahy nebo horizontální oddíly). Tento proces může trvat několik sekund až několik hodin v závislosti na množství dat v počátečním úložišti.

Zrušení zásad příjmu streamování

Zásady příjmu streamování můžete zrušit pomocí Azure Portal nebo programově v jazyce C#.

Azure Portal

1. V Azure Portal přejděte do clusteru Azure Data Explorer a vyberte Dotaz.

2. Pokud chcete zásady příjmu dat streamování z tabulky odstranit, zkopírujte následující příkaz do podokna Dotaz a vyberte Spustit.

   .delete table TestTable policy streamingingestion
   

   

3. V Nastavení vyberte Konfigurace.

4. V podokně Konfigurace vyberte Vypnuto a zakažte příjem dat streamování.

5. Vyberte Uložit.

   

C#

Pokud chcete z tabulky odstranit zásady příjmu streamování, spusťte následující kód:

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

Pokud chcete zakázat ingestování streamování v clusteru, spusťte následující kód:

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

Omezení

• Mapování dat musí být předem vytvořené pro použití při příjmu streamů. Jednotlivé požadavky na příjem streamovaných dat nepoužádá vložená mapování dat.
• U streamovaných dat příjmu dat není možné nastavit značky rozsahu.
• Aktualizujte zásady. Zásady aktualizace mohou odkazovat pouze na nově přijatá data ve zdrojové tabulce, a ne na jiná data nebo tabulky v databázi.
• Pokud je v clusteru, který se používá jako vedoucí pro databáze sledujících, povolený ingestování streamování, musí být ingestování streamování povolené i v následujících clusterech, aby bylo možné sledovat streamovaná data příjmu dat. Totéž platí pro sdílení dat clusteru prostřednictvím Data Share.

Další kroky

• Dotazování dat v Azure Data Explorer