Configurar ingestão de streaming no cluster do Azure Data Explorer

A ingestão de streaming é útil para carregar dados quando você precisa de baixa latência entre a ingestão e a consulta. Considere o uso da ingestão de streaming nos seguintes cenários:

• A latência de menos de um segundo é necessária.
• Para otimizar o processamento operacional de muitas tabelas em que o fluxo de dados em cada tabela é relativamente pequeno (alguns registros por segundo), mas o volume geral de ingestão de dados é alto (milhares de registros por segundo).

Se o fluxo de dados em cada tabela for alto (mais de 4 GB por hora), considere usar a ingestão enfileirada.

Para saber mais sobre métodos de ingestão diferentes, confira a visão geral de ingestão de dados.

Para exemplos de código baseados em versões anteriores do SDK, consulte o artigo arquivado.

Escolha o tipo de ingestão de streaming adequado

Há suporte para dois tipos de ingestão de streaming:

Tipo de ingestão	Descrição
Data connection	As conexões de dados dos Hubs de Eventos, do Hub IoT e da Grade de Eventos podem usar a ingestão de streaming, desde que ela esteja habilitada no nível do cluster. A decisão de usar a assimilação de streaming é feita de acordo com a política de assimilação de streaming configurada na tabela de destino. Para obter informações sobre como gerenciar conexões de dados, consulte Hub de Eventos, Hub IoT e Grade de Eventos.
Ingestão personalizada	A ingestão personalizada exige que você escreva um aplicativo que use uma das bibliotecas de cliente do Azure Data Explorer. Use as informações neste tópico para configurar a ingestão personalizada. Você também pode achar útil o aplicativo de exemplo de ingestão de streaming C?view=azure-data-explorer&preserve-view=true#.

Use a tabela a seguir para ajudar você a escolher o tipo de ingestão apropriado para seu ambiente:

Critério	Data connection	Ingestão personalizada
Atraso de dados entre o início da ingestão e os dados disponíveis para consulta	Atraso maior	Atraso menor
Sobrecarga de desenvolvimento	Configuração rápida e fácil, sem sobrecarga de desenvolvimento	Alta sobrecarga de desenvolvimento para criar um aplicativo, ingerir os dados, tratar erros e garantir a consistência dos dados

Observação

Você pode gerenciar o processo para habilitar e desabilitar a ingestão de streaming em seu cluster usando o portal do Azure ou programaticamente no C#. Se você estiver usando o C# para o seu aplicativo personalizado, talvez seja mais conveniente usar a abordagem programática.

Pré-requisitos

• Uma assinatura do Azure. Criar uma conta gratuita do Azure.

Considerações operacionais e de design

Os principais fatores que podem afetar a ingestão de streaming são:

• Tamanho de cluster e VM: o desempenho e a capacidade da ingestão de streaming escala de acordo com os tamanhos maiores de VM e cluster. O número de solicitações de ingestão simultâneas é limitado a seis por núcleo. Por exemplo, para SKUs de 16 núcleos, como D14 e L16, a carga máxima com suporte é de 96 solicitações de ingestão simultâneas. Para SKUs de dois núcleos, como D11, a carga máxima com suporte é de 12 solicitações de ingestão simultâneas.
• Limite do tamanho dos dados: o limite do tamanho dos dados de uma solicitação de ingestão de streaming é de 4 MB. Isso inclui todos os dados criados para políticas de atualização durante a ingestão.
• Atualizações de esquema: as atualizações de esquema, como criação e modificação de tabelas e mapeamentos de ingestão, podem levar até cinco minutos para o serviço de ingestão de streaming. Para obter mais informações, consulte Ingestão de streaming e alterações de esquema.
• Capacidade do SSD: habilitar a ingestão de streaming em um cluster, mesmo quando os dados não são ingeridos por streaming, ocasiona o uso de parte do disco SSD local dos computadores do cluster para os dados da ingestão de streaming e reduz o armazenamento disponível para o cache de acesso frequente.

Habilitar ingestão de streaming no cluster

Para usar a ingestão de streaming, habilite a funcionalidade em seu cluster e defina uma política de ingestão de streaming. Você pode habilitar a funcionalidade ao criar o cluster ou adicioná-la a um cluster existente.

Aviso

Examine as limitações antes de habilitar a ingestão de streaming.

Habilitar ingestão de streaming ao criar um novo cluster

Você pode habilitar a ingestão de streaming ao criar um cluster usando o portal do Azure ou programaticamente no C#.

Portal

Ao criar um cluster usando as etapas em Criar um cluster e um banco de dados do Azure Data Explorer, na guia Configurações, selecione Ingestão de streaming>Ativada.

C#

Para habilitar a ingestão de streaming ao criar um cluster do Azure Data Explorer, execute o seguinte código:

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

Habilitar ingestão de streaming em um cluster existente

Se você já tem um cluster, habilite a ingestão de streaming usando o portal do Azure ou programaticamente no C#.

Portal

1. No portal do Azure, acesse o cluster do Azure Data Explorer.

2. Em Configurações, selecione Configurações.

3. No painel Configurações, selecione Ativado para habilitar a Ingestão de streaming.

4. Selecione Salvar.

   

C#

Você pode habilitar a ingestão de streaming ao atualizar um cluster existente do 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);
    }
}

Criar uma tabela de destino e definir a política

Crie uma tabela para receber os dados de ingestão de streaming e defina a política relacionada usando o portal do Azure ou programaticamente no C#.

Portal

1. No portal do Azure, navegue até o cluster.

2. Selecione Consulta.

   

3. Para criar a tabela que receberá os dados usando a ingestão de streaming, copie o comando a seguir no painel Consulta e selecione Executar.

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

   

4. Copie um dos comandos a seguir no painel Consulta e selecione Executar. Isso define a política de ingestão de streaming na tabela criada ou no banco de dados que contém a tabela.

   Dica

   Uma política definida no nível do banco de dados se aplica a todas as tabelas existentes e futuras no banco de dados. Quando você habilita a política no nível do banco de dados, não há necessidade de habilitá-la por tabela.

   • Para definir a política na tabela que você criou, use:

     .alter table TestTable policy streamingingestion enable
     

   • Para definir a política no banco de dados que contém a tabela que você criou, use:

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

Criar um aplicativo de ingestão de streaming para ingerir dados para o seu cluster

Crie seu aplicativo para ingerir dados para o seu cluster usando sua linguagem preferida.

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

Go

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

Desabilitar a ingestão de streaming no cluster

Aviso

Desabilitar a ingestão de streaming pode levar algumas horas.

Antes de desabilitar a ingestão de streaming no cluster do Azure Data Explorer, remova a política de ingestão de streaming de todas as tabelas e bancos de dados relevantes. A remoção da política de ingestão de streaming dispara a reorganização de dados no cluster do Azure Data Explorer. Os dados de ingestão de streaming são movidos do armazenamento inicial para o armazenamento permanente no repositório de coluna (extensões ou fragmentos). Esse processo pode levar entre alguns segundos a algumas horas, dependendo da quantidade de dados no armazenamento inicial.

Remover a política de ingestão de streaming

Você pode remover a política de ingestão de streaming usando o portal do Azure ou programaticamente no C#.

Portal

1. No portal do Azure, acesse o cluster do Azure Data Explorer e selecione Consulta.

2. Para soltar a política de ingestão de streaming na tabela, copie o comando a seguir no painel Consulta e selecione Executar.

   .delete table TestTable policy streamingingestion
   

   

3. Em Configurações, selecione Configurações.

4. No painel Configurações, selecione Desativado para desabilitar a Ingestão de streaming.

5. Selecione Salvar.

   

C#

Para soltar a política de ingestão de streaming da tabela, execute o seguinte código:

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

Para desabilitar a ingestão de streaming no cluster, execute o seguinte código:

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

Limitações

• Os Mapeamentos de dados devem ser pré-criados para uso na ingestão de streaming. Solicitações de ingestão de streaming individuais não acomodam mapeamentos de dados em linha.
• As marcas de extensão não podem ser definidas nos dados de ingestão de streaming.
• Atualizar política. A política de atualização pode mencionar apenas os dados recém-ingeridos na tabela de origem e não quaisquer outros dados ou tabelas no banco de dados.
• Quando uma política de atualização com uma política transacional falha, as novas tentativas retornam à ingestão em lote.
• Se a ingestão de streaming estiver habilitada em um cluster usado como líder para bancos de dados de acompanhamento, a ingestão de streaming deverá ser habilitada nos clusters a seguir, bem como seguir os dados de ingestão de streaming. O mesmo se aplica se os dados do cluster são compartilhados por meio do Data Share.

Conteúdo relacionado

• Consultar dados no Azure Data Explorer