Configurar a ingestão de dados da transmissão em fluxo no cluster do Azure Data Explorer

A ingestão de transmissão em fluxo é útil para carregar dados quando precisa de baixa latência entre a ingestão e a consulta. Considere utilizar a ingestão de transmissão em fluxo nos seguintes cenários:

• É necessária latência inferior a um segundo.
• Para otimizar o processamento operacional de muitas tabelas em que o fluxo de dados para cada tabela é relativamente pequeno (alguns registos por segundo), mas o volume global de ingestão de dados é elevado (milhares de registos por segundo).

Se o fluxo de dados para cada tabela for elevado (mais de 4 GB por hora), considere utilizar a ingestão em fila.

Para saber mais sobre diferentes métodos de ingestão, veja Descrição geral da ingestão de dados.

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

Escolha o tipo de ingestão de transmissão em fluxo adequado

São suportados dois tipos de ingestão de transmissão em fluxo:

Tipo de ingestão	Description
Ligação de dados	Os Hubs de Eventos, Hub IoT e ligações de dados do Event Grid podem utilizar a ingestão de transmissão em fluxo, desde que esteja ativada ao nível do cluster. A decisão de utilizar a ingestão de transmissão em fluxo é tomada de acordo com a política de ingestão de transmissão em fluxo configurada na tabela de destino. Para obter informações sobre como gerir ligações de dados, veja Hub de Eventos, Hub IoT e Event Grid.
Ingestão personalizada	A ingestão personalizada requer que escreva uma aplicação que utilize uma das bibliotecas de cliente do Azure Data Explorer. Utilize as informações neste tópico para configurar a ingestão personalizada. Também pode considerar útil a aplicação de exemplo de ingestão de transmissão em fluxo C# .

Utilize a seguinte tabela para o ajudar a escolher o tipo de ingestão adequado para o seu ambiente:

Critério	Ligação de dados	Ingestão Personalizada
Atraso de dados entre o início da ingestão e os dados disponíveis para consulta	Atraso mais longo	Atraso mais curto
Sobrecarga de desenvolvimento	Configuração rápida e fácil, sem sobrecarga de desenvolvimento	Sobrecarga de desenvolvimento elevada para criar uma aplicação ingerir os dados, lidar com erros e garantir a consistência dos dados

Nota

Pode gerir o processo para ativar e desativar a ingestão de transmissão em fluxo no cluster com o portal do Azure ou programaticamente em C#. Se estiver a utilizar C# para a sua aplicação personalizada, poderá considerar mais conveniente utilizar a abordagem programática.

Pré-requisitos

• Uma subscrição do Azure. Crie uma conta gratuita do Azure.

Considerações de desempenho e operacionais

Os principais contribuintes que podem afetar a ingestão de transmissão em fluxo são:

• Tamanho da VM e do cluster: o desempenho da ingestão de transmissão em fluxo e as escalas de capacidade com tamanhos de VM e cluster aumentados. O número de pedidos de ingestão simultâneos está limitado a seis por núcleo. Por exemplo, para SKUs de 16 núcleos, como D14 e L16, a carga máxima suportada é de 96 pedidos de ingestão simultâneos. Para dois SKUs principais, como D11, a carga máxima suportada é de 12 pedidos de ingestão simultâneos.
• Limite de tamanho dos dados: o limite de tamanho dos dados para um pedido de ingestão de transmissão em fluxo é de 4 MB. Isto inclui quaisquer dados criados para políticas de atualização durante a ingestão.
• Atualizações de esquema: as atualizações de esquema, como a criação e modificação de tabelas e mapeamentos de ingestão, podem demorar até cinco minutos para o serviço de ingestão de transmissão em fluxo. Para obter mais informações, veja Ingestão de transmissão em fluxo e alterações de esquema.
• Capacidade SSD: ativar a ingestão de transmissão em fluxo num cluster, mesmo quando os dados não são ingeridos através da transmissão em fluxo, utiliza parte do disco SSD local das máquinas de cluster para transmitir dados de ingestão de dados e reduz o armazenamento disponível para cache frequente.

Ativar a ingestão de transmissão em fluxo no cluster

Antes de poder utilizar a ingestão de transmissão em fluxo, tem de ativar a capacidade no cluster e definir uma política de ingestão de transmissão em fluxo. Pode ativar a capacidade ao criar o cluster ou adicioná-lo a um cluster existente.

Aviso

Reveja as limitações antes de ativar a ingestão de transmissão em fluxo.

Ativar a ingestão de transmissão em fluxo ao criar um novo cluster

Pode ativar a ingestão de transmissão em fluxo ao criar um novo cluster com o portal do Azure ou programaticamente em C#.

Portal

Ao criar um cluster com os passos em Criar um cluster e uma base de dados do Azure Data Explorer, no separador Configurações, selecione Ingestão de transmissão> em fluxoAtivada.

C#

Para ativar a ingestão de transmissão em fluxo ao criar um novo 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);
    }
}

Ativar a ingestão de transmissão em fluxo num cluster existente

Se tiver um cluster existente, pode ativar a ingestão de transmissão em fluxo com o portal do Azure ou programaticamente em C#.

Portal

1. No portal do Azure, aceda ao cluster do Azure Data Explorer.

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

3. No painel Configurações , selecione Ativado para ativar a ingestão de transmissão em fluxo.

4. Selecione Guardar.

   

C#

Pode ativar a ingestão de transmissão em fluxo ao atualizar um cluster do Azure Data Explorer existente.

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 transmissão em fluxo e definir a política relacionada com o portal do Azure ou programaticamente em C#.

Portal

1. No portal do Azure, navegue para o cluster.

2. Selecione Consulta.

   

3. Para criar a tabela que irá receber os dados através da ingestão de transmissão em fluxo, copie o seguinte comando para o painel Consulta e selecioneExecutar.

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

   

4. Copie um dos seguintes comandos para o painel Consulta e selecioneExecutar. Isto define a política de ingestão de transmissão em fluxo na tabela que criou ou na base de dados que contém a tabela.

   Dica

   Uma política definida ao nível da base de dados aplica-se a todas as tabelas existentes e futuras na base de dados. Quando ativa a política ao nível da base de dados, não é necessário ativá-la por tabela.

   • Para definir a política na tabela que criou, utilize:

     .alter table TestTable policy streamingingestion enable
     

   • Para definir a política na base de dados que contém a tabela que criou, utilize:

     .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 uma aplicação de ingestão de transmissão em fluxo para ingerir dados no cluster

Crie a sua aplicação para ingerir dados no cluster com o seu idioma preferido.

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

Ir

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

Desativar a ingestão de transmissão em fluxo no cluster

Aviso

A desativação da ingestão de transmissão em fluxo pode demorar algumas horas.

Antes de desativar a ingestão de transmissão em fluxo no cluster do Azure Data Explorer, remova a política de ingestão de transmissão em fluxo de todas as tabelas e bases de dados relevantes. A remoção da política de ingestão de transmissão em fluxo aciona a reorganização de dados no cluster do Azure Data Explorer. Os dados de ingestão de transmissão em fluxo são movidos do armazenamento inicial para o armazenamento permanente no arquivo de colunas (extensões ou partições horizontais). Este processo pode demorar entre alguns segundos a algumas horas, dependendo da quantidade de dados no armazenamento inicial.

Remover a política de ingestão de transmissão em fluxo

Pode remover a política de ingestão de transmissão em fluxo com o portal do Azure ou programaticamente em C#.

Portal

1. No portal do Azure, aceda ao cluster do Azure Data Explorer e selecione Consulta.

2. Para remover a política de ingestão de transmissão em fluxo da tabela, copie o seguinte comando para o painel Consulta e selecioneExecutar.

   .delete table TestTable policy streamingingestion
   

   

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

4. No painel Configurações , selecione Desativar para desativar a ingestão de transmissão em fluxo.

5. Selecione Guardar.

   

C#

Para remover a política de ingestão de transmissão em fluxo 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 desativar a ingestão de transmissão em fluxo 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 têm de ser pré-criados para utilização na ingestão de transmissão em fluxo. Os pedidos individuais de ingestão de transmissão em fluxo não acomodam mapeamentos de dados inline.
• As etiquetas de extensão não podem ser definidas nos dados de ingestão de transmissão em fluxo.
• Atualizar política. A política de atualização só pode referenciar os dados ingeridos recentemente na tabela de origem e não quaisquer outros dados ou tabelas na base de dados.
• Se a ingestão de transmissão em fluxo estiver ativada num cluster utilizado como líder para bases de dados de seguidores, a ingestão de transmissão em fluxo tem de ser ativada nos seguintes clusters, bem como para seguir os dados de ingestão de transmissão em fluxo. O mesmo se aplica se os dados do cluster são partilhados através de Data Share.

Conteúdo relacionado

• Consultar dados no Azure Data Explorer