Streamelési adatbetöltés konfigurálása az Azure Data Explorer-fürtön

A streambetöltés akkor hasznos az adatok betöltéséhez, ha kis késésre van szüksége a betöltés és a lekérdezés között. Fontolja meg a streambetöltés használatát a következő esetekben:

• Egy másodpercnél rövidebb késésre van szükség.
• Számos olyan tábla működési feldolgozásának optimalizálása érdekében, ahol az adatok adatfolyama az egyes táblákba viszonylag kicsi (másodpercenként néhány rekord), de a teljes adatbetöltési mennyiség magas (több ezer rekord másodpercenként).

Ha az adatok adatfolyama az egyes táblákba magas (óránként 4 GB-nál nagyobb), fontolja meg az üzenetsoros betöltés használatát.

A különböző betöltési módszerekkel kapcsolatos további információkért lásd az adatbetöltés áttekintését.

A korábbi SDK-verziókon alapuló kódmintákért tekintse meg az archivált cikket.

Válassza ki a megfelelő streamelési betöltési típust

Két streamelési betöltési típus támogatott:

Betöltési típus	Description
Adatkapcsolat	Az Event Hubs, a IoT Hub és az Event Grid-adatkapcsolatok streambetöltést használhatnak, feltéve, hogy engedélyezve van a fürt szintjén. A streambetöltés használatára vonatkozó döntés a céltáblán konfigurált streamelési betöltési szabályzatnak megfelelően történik. Az adatkapcsolatok kezelésével kapcsolatos információkért lásd: Event Hub, IoT Hub és Event Grid.
Egyéni betöltés	Az egyéni betöltéshez olyan alkalmazást kell írnia, amely az Azure Data Explorer ügyfélkódtárak egyikét használja. Az ebben a témakörben található információk segítségével konfigurálhatja az egyéni betöltési adatokat. A C#-streamelési mintaalkalmazás is hasznos lehet.

Az alábbi táblázat segítségével kiválaszthatja a környezetének megfelelő betöltési típust:

Kritérium	Adatkapcsolat	Egyéni betöltés
Adatbetöltés kezdeményezése és a lekérdezéshez elérhető adatok közötti adatkésés	Hosszabb késleltetés	Rövidebb késleltetés
Fejlesztési többletterhelés	Gyors és egyszerű beállítás, nincs fejlesztési többletterhelés	Nagy fejlesztési terhelés az adatok betöltését, a hibák kezelését és az adatkonzisztenciát biztosító alkalmazások létrehozásához

Megjegyzés

A streambetöltés engedélyezésének és letiltásának folyamatát a Azure Portal vagy programozott módon c# nyelven kezelheti a fürtön. Ha az egyéni alkalmazáshoz C#-ot használ, a programozott megközelítéssel kényelmesebbnek találhatja.

Előfeltételek

• Azure-előfizetés. Hozzon létre egy ingyenes Azure-fiókot.

Teljesítménnyel és működésével kapcsolatos szempontok

A streambetöltést befolyásoló fő közreműködők a következők:

• Virtuális gép és fürt mérete: A streamelési betöltési teljesítmény és a kapacitás skálázható a virtuális gép és a fürt méretének növelésével. Az egyidejű betöltési kérelmek száma magonként legfeljebb hat lehet. Például a 16 magos termékváltozatok, például a D14 és az L16 esetében a maximálisan támogatott terhelés 96 egyidejű betöltési kérés. Két alapvető termékváltozat , például a D11 esetében a maximálisan támogatott terhelés 12 egyidejű betöltési kérelem.
• Adatméretkorlát: A streamelési betöltési kérelem adatméretkorlátja 4 MB. Ide tartoznak a frissítési szabályzatokhoz a betöltés során létrehozott adatok.
• Sémafrissítések: A sémafrissítések, például a táblák és a betöltési leképezések létrehozása és módosítása akár öt percet is igénybe vehet a streamelési szolgáltatás számára. További információ: Streambetöltés és sémamódosítások.
• SSD-kapacitás: A streamelési adatbetöltés engedélyezése egy fürtön még akkor is, ha az adatok nem streameléssel vannak betöltve, a fürtgépek helyi SSD-lemezének egy részét használja a betöltési adatok streamelésére, és csökkenti a gyorsgyorsítótár számára elérhető tárhelyet.

Streambetöltés engedélyezése a fürtön

Mielőtt használhatja a streambetöltést, engedélyeznie kell a képességet a fürtön, és meg kell határoznia egy streamelési betöltési szabályzatot. Engedélyezheti a képességet a fürt létrehozásakor, vagy hozzáadhatja egy meglévő fürthöz.

Figyelmeztetés

A streambetöltés engedélyezése előtt tekintse át a korlátozásokat .

Streambetöltés engedélyezése új fürt létrehozásakor

Új fürt létrehozásakor engedélyezheti a streambetöltést a Azure Portal vagy programozott módon C# nyelven.

Portál

Fürt létrehozásakor az Azure Data Explorer-fürt és -adatbázis létrehozása lépésekkel, a Konfigurációk lapon válassza a Streambetöltés>bekapcsolva lehetőséget.

C#

Ha engedélyezni szeretné a streambetöltést egy új Azure Data Explorer-fürt létrehozásakor, futtassa a következő kódot:

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

Streambetöltés engedélyezése meglévő fürtön

Ha már rendelkezik fürtel, engedélyezheti a streambetöltést a Azure Portal vagy programozott módon C# nyelven.

Portál

1. A Azure Portal lépjen az Azure Data Explorer-fürtre.

2. A Beállítások területen válassza a Konfigurációk lehetőséget.

3. A Konfigurációk panelen válassza a Be lehetőséget a streambetöltés engedélyezéséhez.

4. Kattintson a Mentés gombra.

   

C#

Engedélyezheti a streambetöltést egy meglévő Azure Data Explorer-fürt frissítésekor.

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

Céltábla létrehozása és a szabályzat meghatározása

Hozzon létre egy táblát a streambetöltési adatok fogadásához, és definiálja a kapcsolódó szabályzatot a Azure Portal vagy programozott módon C# nyelven.

Portál

1. A Azure Portal keresse meg a fürtöt.

2. Válassza a Lekérdezés lehetőséget.

   

3. Ha streambetöltéssel szeretné létrehozni az adatokat fogadó táblát, másolja a következő parancsot a Lekérdezés panelre , és válassza a Futtatás lehetőséget.

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

   

4. Másolja az alábbi parancsok egyikét a Lekérdezés panelre , és válassza a Futtatás lehetőséget. Ez határozza meg a streamelési betöltési szabályzatot a létrehozott táblában vagy a táblát tartalmazó adatbázisban.

   Tipp

   Az adatbázis szintjén definiált szabályzat az adatbázis összes meglévő és jövőbeli táblájára vonatkozik. Ha az adatbázis szintjén engedélyezi a szabályzatot, nem szükséges táblánként engedélyeznie.

   • A létrehozott táblára vonatkozó szabályzat meghatározásához használja a következőt:

     .alter table TestTable policy streamingingestion enable
     

   • A létrehozott táblát tartalmazó adatbázis szabályzatának meghatározásához használja a következőt:

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

Streamelési adatbetöltési alkalmazás létrehozása a fürtbe való adatbetöltéshez

Hozza létre az alkalmazást az adatok fürtbe való betöltéséhez az előnyben részesített nyelv használatával.

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

Ugrás

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

Streambetöltés letiltása a fürtön

Figyelmeztetés

A streambetöltés letiltása eltarthat néhány óráig.

Mielőtt letiltja a streambetöltést az Azure Data Explorer-fürtön, elvetheti a streamelési betöltési szabályzatot az összes releváns táblából és adatbázisból. A streamelési betöltési szabályzat eltávolítása adatáthelyezést vált ki az Azure Data Explorer-fürtön belül. A streamelési betöltési adatok átkerülnek a kezdeti tárolóból az oszloptároló állandó tárolójába (kiterjedések vagy szegmensek). Ez a folyamat a kezdeti tárolóban lévő adatok mennyiségétől függően néhány másodperctől néhány óráig is eltarthat.

A streamelési betöltési szabályzat elvetése

A streamelési betöltési szabályzatot elvetheti a Azure Portal vagy programozott módon C# nyelven.

Portál

1. A Azure Portal lépjen az Azure Data Explorer-fürtre, és válassza a Lekérdezés lehetőséget.

2. Ha el szeretné dobni a streamelési betöltési szabályzatot a táblából, másolja a következő parancsot a Lekérdezés panelre , és válassza a Futtatás lehetőséget.

   .delete table TestTable policy streamingingestion
   

   

3. A Beállítások területen válassza a Konfigurációk lehetőséget.

4. A Konfigurációk panelen válassza a Ki lehetőséget a streambetöltés letiltásához.

5. Kattintson a Mentés gombra.

   

C#

A streamelési betöltési szabályzat táblából való elvetéséhez futtassa a következő kódot:

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

Ha le szeretné tiltani a streambetöltést a fürtön, futtassa a következő kódot:

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

Korlátozások

• Az adatleképezéseketelőre létre kell hozni a streambetöltéshez. Az egyes streamelési betöltési kérések nem fogadják el a beágyazott adatleképezéseket.
• A bővítménycímkék nem állíthatók be a streamelési betöltési adatokon.
• Szabályzat frissítése. A frissítési szabályzat csak a forrástábla újonnan betöltött adataira hivatkozhat, az adatbázisban lévő többi adatra vagy táblára nem.
• Ha a streambetöltés engedélyezve van egy olyan fürtön, amely vezető szerepet tölt be a követő adatbázisokban, a streambetöltést a következő fürtökön is engedélyezni kell a streamelési betöltési adatok követéséhez. Ugyanez vonatkozik arra is, hogy a fürt adatai Data Share keresztül legyenek-e megosztva.

Kapcsolódó tartalom

• Adatok lekérdezése az Azure Data Explorer