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

Streaming ingestace je užitečná pro načítání dat, když potřebujete nízkou latenci mezi ingestací a dotazem. Zvažte použití příjmu streamovaných dat v následujících scénářích:

• Je vyžadována latence menší než sekunda.
• Optimalizace provozního zpracování mnoha tabulek, kde je datový proud 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 do každé tabulky vysoký (přes 4 GB za hodinu), zvažte použití frontovaného ingestování dat.

Další informace o různých metodách příjmu dat najdete v přehledu příjmu dat.

Ukázky kódu založené na předchozích verzích sady SDK najdete v archivovaného článku.

Zvolte odpovídající typ příjmu streamovaných dat.

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

Typ příjmu dat	Popis
Datové připojení	Služby Event Hubs, IoT Hub a datové připojení Event Grid mohou využívat streamovací příjem, pokud je povoleno na úrovni clusteru. Rozhodnutí o použití příjmu streamovaných dat se provádí podle zásad příjmu streamování nakonfigurovaných v cílové tabulce. Informace o správě datových připojení najdete v tématu Event Hub, IoT Hub a Event Grid.
Vlastní ingestování dat	Vlastní příjem dat vyžaduje, abyste napsali aplikaci, která používá některou z klientských knihoven Azure Data Exploreru. Informace v tomto článku slouží ke konfiguraci přizpůsobeného příjmu dat. Můžete také najít užitečnou ukázkovou aplikaci pro streamování příjmu dat na C?view=azure-data-explorer&preserve-view=true#.

Následující tabulka vám pomůže zvolit 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žijní náklady na vývoj	Rychlé a snadné nastavení bez režijních nákladů na vývoj	Vysoká režie při vývoji aplikace pro přijímání dat, zpracování chyb a zajištění konzistence dat.

Poznámka:

Proces můžete spravovat, abyste v clusteru povolili a zakázali příjem dat streamování pomocí webu Azure Portal nebo programově v jazyce C#. Pokud používáte jazyk C# pro vlastní aplikaci, může být vhodnější použít programový přístup.

Požadavky

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

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

Hlavní faktory, které mohou ovlivnit zpracování streamovaných dat, jsou:

• Velikost virtuálního počítače a clusteru: Výkon streamovaného příjmu dat a kapacita se škáluje se zvětšujícími se 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 16 jader SKU, například D14 a L16, je maximální podporovaná zátěž 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.
• Omezení velikosti dat: Limit velikosti dat pro požadavek na příjem dat 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í ingestace, můžou trvat až pět minut pro službu pro streamingovou ingestaci. Další informace najdete v tématu Streamování příjmu dat a změny schématu.
• Kapacita SSD: Povolení streamovacího příjmu dat v clusteru, i když data nejsou přijímána prostřednictvím streamování, využívá část místního disku SSD počítačů clusteru pro data ze streamovacího příjmu a snižuje dostupné úložiště pro aktivní mezipaměť.
• Kurzory databáze: Při streamovacím příjmu dat se může aktualizace kurzoru databáze opožďovat až o 60 sekund za dostupností dat. Toto zpoždění vzniká z asynchronních procesů uzavírání na pozadí, které převádějí data z vyrovnávací paměti streamování na trvalé úseky úložiště ve sloupcích, během nichž se aktualizuje kurzor (používaný pro přírůstkové zpracování, průběžné exporty nebo materializovaná zobrazení). Pokud vaše úloha vyžaduje okamžitou konzistenci kurzoru pro sémantiku "pouze jednou," zvažte místo toho použití příjmu pomocí front nebo zohledněte tuto potenciální prodlevu ve vaší aplikační logice.

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

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

Varování

Před povolením příjmu streamovaných dat zkontrolujte omezení .

Povolit streamovací příjem při vytváření nového clusteru

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

Portal

Při vytváření clusteru podle kroků v Vytvoření clusteru a databáze Azure Data Exploreru, na kartě Konfigurace vyberte Streamovací příjem >zapnuto.

C#

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

using System.Threading.Tasks;
using Azure;
using Azure.Core;
using Azure.Identity; // Required package Azure.Identity
using Azure.ResourceManager;
using Azure.ResourceManager.Kusto; // Required package Azure.ResourceManager.Kusto
using Azure.ResourceManager.Kusto.Models;
namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        var subscriptionId = "<subscriptionId>";
        var credentials = new ClientSecretCredential(appTenant, appId, appKey);
        var resourceManagementClient = new ArmClient(credentials, subscriptionId);
        var resourceGroupName = "<resourceGroupName>";
        var clusterName = "<clusterName>";
        
        var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
        var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
        var clusters = resourceGroup.GetKustoClusters();
        
        var location = new AzureLocation("<location>");
        var skuName = new KustoSkuName("<skuName>");
        var skuTier = new KustoSkuTier("<skuTier>");
        var clusterData = new KustoClusterData(location, new KustoSku(skuName, skuTier)) { IsStreamingIngestEnabled = true };
        await clusters.CreateOrUpdateAsync(WaitUntil.Completed, clusterName, clusterData);
    }
}

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

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

Portal

1. Na webu Azure Portal přejděte do clusteru Azure Data Exploreru.

2. V Nastavení vyberte Konfigurace.

3. V podokně Konfigurace vyberte Zapnuto a povolte streamovací příjem.

4. Zvolte Uložit.

   

C#

Můžete povolit streamové zpracování dat při aktualizaci existujícího clusteru Azure Data Explorer.

using System.Threading.Tasks;
using Azure;
using Azure.Identity; // Required package Azure.Identity
using Azure.ResourceManager;
using Azure.ResourceManager.Kusto; // Required package Azure.ResourceManager.Kusto
using Azure.ResourceManager.Kusto.Models;
namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        var subscriptionId = "<subscriptionId>";
        var credentials = new ClientSecretCredential(appTenant, appId, appKey);
        var resourceManagementClient = new ArmClient(credentials, subscriptionId);
        var resourceGroupName = "<resourceGroupName>";
        var clusterName = "<clusterName>";
        var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
        var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
        var cluster = (await resourceGroup.GetKustoClusterAsync(clusterName)).Value;
        var clusterPatch = new KustoClusterPatch(cluster.Data.Location) { IsStreamingIngestEnabled = true };
        await cluster.UpdateAsync(WaitUntil.Completed, clusterPatch);
    }
}

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

Vytvořte tabulku pro příjem streamovaných dat a definujte související zásady použitím Azure portálu nebo programováním v C#.

Portal

1. Na webu Azure Portal přejděte do svého clusteru.

2. Vyberte Dotaz.

   

3. Pokud chcete vytvořit tabulku, která bude přijímat data prostřednictvím příjmu streamovaných dat, zkopírujte do podokna Dotaz následující příkaz 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 streamovacího příjmu dat na tabulce, kterou jste vytvořili, nebo na databázi, která tuto tabulku obsahuje.

   Návod

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

   • K definování zásad v tabulce, kterou jste vytvořili, použijte:

     .alter table TestTable policy streamingingestion enable
     

   • K definování zásad databáze obsahující tabulku, kterou jste vytvořili, použijte:

     .alter database StreamingTestDb policy streamingingestion enable
     

   

C#

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řte aplikaci pro streamovací příjem dat do vašeho clusteru.

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

Poznámka:

Informace o příjmu dat ve frontě najdete v pokynech k vytvoření aplikace pro získání dat pomocí příjmu dat ve frontě.

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

Spustit

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

func ingest() {
    clusterPath := "https://<clusterName>.<region>.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;
    }
}

Zakázání příjmu streamovaných dat v clusteru

Varování

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

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

Zrušení zásad pro streamovací příjem

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

Portal

1. Na webu Azure Portal přejděte do clusteru Azure Data Exploreru a vyberte Dotaz.

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

   .delete table TestTable policy streamingingestion
   

   

3. V Nastavení vyberte Konfigurace.

4. V podokně Konfigurace vyberte Vypnuto a zakažte streamovací ingest.

5. Zvolte Uložit.

   

C#

Pokud chcete z tabulky odstranit zásady streamovaného příjmu, 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 v clusteru zakázat příjem streamu, spusťte následující kód:

using System.Threading.Tasks;
using Azure;
using Azure.Identity; // Required package Azure.Identity
using Azure.ResourceManager;
using Azure.ResourceManager.Kusto; // Required package Azure.ResourceManager.Kusto
using Azure.ResourceManager.Kusto.Models;
namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        var subscriptionId = "<subscriptionId>";
        var credentials = new ClientSecretCredential(appTenant, appId, appKey);
        var resourceManagementClient = new ArmClient(credentials, subscriptionId);
        var resourceGroupName = "<resourceGroupName>";
        var clusterName = "<clusterName>";
        var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
        var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
        var cluster = (await resourceGroup.GetKustoClusterAsync(clusterName)).Value;
        var clusterPatch = new KustoClusterPatch(cluster.Data.Location) { IsStreamingIngestEnabled = false };
        await cluster.UpdateAsync(WaitUntil.Completed, clusterPatch);
    }
}

Omezení

• Mapování dat musí být předem vytvořené pro použití při streamovaném příjmu dat. Jednotlivé požadavky na příjem datového streamování nepodporují mapování dat inline.
• U streamovaných dat příjmu dat není možné nastavit značky rozsahu.
• Aktualizace zásad
  • Zásady aktualizace můžou odkazovat pouze na nově přijatá data ve zdrojové tabulce, a ne na žádná jiná data ani tabulky v databázi.
  • Modul plug-in Pythonu se nepodporuje.
  • Když se zásada aktualizace s transakčními zásadami nezdaří, opakování se vrátí do dávkového příjmu dat.
  • U kaskádových zásad aktualizací , které zahrnují join operátora, je nutné zakázat příjem streamů u všech nadřazených tabulek. Zvažte například kaskádové zásady aktualizace, kde tabulka1 aktualizuje tabulku 2, tabulka2 aktualizuje tabulku 3 a tabulku 3 aktualizuje tabulku 4. Pokud zásady aktualizace tabulky Table4 zahrnují spojení, musíte zakázat streamové ingestování u tabulek Table1, Table2 a Table3.
• Pokud je v clusteru, který slouží jako vedoucí pro sledující databáze, povolený streamovací příjem, musí být streamovací příjem povolený také na následujících clusterech, aby mohly sledovat data streamovaného příjmu. Totéž platí pro případ, kdy jsou data clusteru sdílena prostřednictvím Data Share.

Související obsah

• Dotazování na data s využitím Azure Data Exploreru