Configurar a ingestão de transmissão em fluxo no conjunto de Azure Synapse Data Explorer (Pré-visualização)

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 de lotes.

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

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	Descrição
Hub de Eventos ou Hub IoT	Os hubs são configurados como origens de dados de transmissão em fluxo de tabelas. Para obter informações sobre como configurar estas definições, veja Hub de Eventos.
Ingestão personalizada	A ingestão personalizada requer que escreva uma aplicação que utilize uma das bibliotecas de cliente Azure Synapse 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	Hub de Eventos/Hub IoT	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

A ingestão de dados de um Hub de Eventos em conjuntos de Data Explorer não funcionará se a área de trabalho do Synapse utilizar uma rede virtual gerida com a proteção de exfiltração de dados ativada.

Pré-requisitos

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

• Criar um conjunto de Data Explorer com Synapse Studio ou o portal do Azure

• Criar uma base de dados Data Explorer.

  1. Em Synapse Studio, no painel esquerdo, selecione Dados.

  2. Selecione + (Adicionar novo recurso) >Data Explorer conjunto e utilize as seguintes informações:

     Definição	Valor sugerido	Descrição
     Nome do conjunto	contosodataexplorer	O nome do conjunto de Data Explorer a utilizar
     Name	TestDatabase	O nome da base de dados tem de ser exclusivo dentro do cluster.
     Período de retenção predefinido	365	O intervalo de tempo (em dias) para o qual é garantido que os dados são mantidos disponíveis para consulta. O intervalo de tempo é medido desde o momento em que os dados são ingeridos.
     Período de cache predefinido	31	O intervalo de tempo (em dias) para manter os dados consultados frequentemente disponíveis no armazenamento SSD ou na RAM, em vez de no armazenamento de longo prazo.

  3. Selecione Criar para criar a base de dados. Normalmente, a criação demora menos de um minuto.

• Obtenha os pontos finais de Consulta e Ingestão de Dados.

  1. Em Synapse Studio, no painel esquerdo, selecione Gerir>Data Explorer conjuntos.

  2. Selecione o conjunto de Data Explorer que pretende utilizar para ver os detalhes.

     

  3. Anote os pontos finais de Consulta e Ingestão de Dados. Utilize o ponto final de Consulta como cluster ao configurar ligações ao conjunto de Data Explorer. Ao configurar SDKs para ingestão de dados, utilize o ponto final de ingestão de dados.

     

Considerações de desempenho e operacionais

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

• Especificação de Computação: o desempenho da ingestão de transmissão em fluxo e as escalas de capacidade com tamanhos de conjunto de Data Explorer aumentados. O número de pedidos de ingestão simultâneos está limitado a seis por núcleo. Por exemplo, para o tipo de carga de trabalho de 16 núcleos, como Otimizado para Computação (Grande) e Otimizado para Armazenamento (Grande), a carga máxima suportada é de 96 pedidos de ingestão simultâneos. Para dois tipos de carga de trabalho principais, como Otimizado para Computação (Extra Pequeno), 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.
• 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 conjunto de Data Explorer, 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 do conjunto de Data Explorer para dados de ingestão de transmissão em fluxo e reduz o armazenamento disponível para cache frequente.

Ativar a ingestão de transmissão em fluxo no conjunto de Data Explorer

Antes de poder utilizar a ingestão de transmissão em fluxo, tem de ativar a capacidade no conjunto de Data Explorer e definir uma política de ingestão de transmissão em fluxo. Pode ativar a capacidade ao criar o conjunto de Data Explorer ou adicioná-la a um conjunto de Data Explorer 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 conjunto de Data Explorer

Pode ativar a ingestão de transmissão em fluxo ao criar um novo conjunto de Data Explorer com o Azure Synapse Studio ou o portal do Azure.

Studio

Ao criar um conjunto de Data Explorer com os passos em Criar um conjunto de Data Explorer com Synapse Studio, no separador Definições adicionais, selecione Ingestão de transmissão> em fluxoAtivada.

Portal

Ao criar um conjunto de Data Explorer com os passos em Criar um conjunto de Data Explorer com o portal do Azure, no separador Definições adicionais, selecione Ingestão de transmissão> em fluxoAtivada.

Ativar a ingestão de transmissão em fluxo num conjunto de Data Explorer existente

Se tiver um conjunto de Data Explorer existente, pode ativar a ingestão de transmissão em fluxo com o portal do Azure.

1. No portal do Azure, aceda ao conjunto de 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.

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 Azure Synapse Studio ou o portal do Azure.

Studio

1. No Synapse Studio, no painel esquerdo, selecione Desenvolver.

2. Em Scripts KQL, selecione + (Adicionar novo recurso) >script KQL. No painel do lado direito, pode atribuir um nome ao script.

3. No menu Ligar a , selecione contosodataexplorer.

4. No menu Utilizar base de dados , selecione TestDatabase.

5. Cole o seguinte comando e selecione Executar para criar a tabela.

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

6. 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.

   • 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
     

Portal

1. No portal do Azure Synapse Analytics, navegue para o conjunto de Data Explorer.

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.

   • 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
     

Criar uma aplicação de ingestão de transmissão em fluxo para ingerir dados no conjunto de Data Explorer

Crie a sua aplicação para ingerir dados no seu conjunto de Data Explorer com o seu idioma preferido. Para a variável poolPath , utilize o ponto final de Consulta de que anota nos Pré-requisitos.

C#

using Kusto.Data;
using Kusto.Ingest;
using System.IO;
using Kusto.Data.Common;

namespace StreamingIngestion
{
    class Program
    {
        static void Main(string[] args)
        {
            string poolPath = "https://<Poolname>.<WorkspaceName>.kusto.windows.net";
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string dbName = "<dbName>";
            string tableName = "<tableName>";

            // Create Kusto connection string with App Authentication
            var csb =
                new KustoConnectionStringBuilder(poolPath)
                    .WithAadApplicationKeyAuthentication(
                        applicationClientId: appId,
                        applicationKey: appKey,
                        authority: appTenant
                    );

            // Create a disposable client that will execute the ingestion
            using (IKustoIngestClient client = KustoIngestFactory.CreateStreamingIngestClient(csb))
            {
                // Initialize client properties
                var ingestionProperties =
                    new KustoIngestionProperties(
                        databaseName: dbName,
                        tableName: tableName
                    );

                // Ingest from a compressed file
                var fileStream = File.Open("MyFile.gz", FileMode.Open);
                // Create source options
                var sourceOptions = new StreamSourceOptions()
                {
                    CompressionType = DataSourceCompressionType.GZip,
                };
                // Ingest from stream
                var status = client.IngestFromStreamAsync(fileStream, ingestionProperties, sourceOptions).GetAwaiter().GetResult();
            }
        }
    }
}

Python

from Azure Synapse Analytics.kusto.data import KustoConnectionStringBuilder

from Azure Synapse Analytics.kusto.ingest import (
    IngestionProperties,
    DataFormat,
    KustoStreamingIngestClient
)

poolPath = "https://<Poolname>.<WorkspaceName>.kusto.windows.net"
appId = "<appId>"
appKey = "<appKey>"
appTenant = "<appTenant>"
dbName = "<dbName>"
tableName = "<tableName>"

csb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
    poolPath,
    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 poolPath = "https://<Poolname>.<WorkspaceName>.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(
        poolPath,
        appId,
        appKey,
        appTenant
    ),
    ingestionProperties
);

// Automatically detects gz format
await client.ingestFromFile("MyFile.gz", ingestionProperties);

Ir

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

func ingest() {
    poolPath := "https://<Poolname>.<WorkspaceName>.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 Synapse Analytics/go-autorest/autorest/Azure Synapse Analytics/auth"
    authorizer := kusto.Authorization{
        Config: auth.NewClientCredentialsConfig(appId, appKey, appTenant),
    }

    // Create a client
    client, err := kusto.New(poolPath, 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 Synapse Analytics.kusto.data.auth.ConnectionStringBuilder;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.IngestClient;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.IngestClientFactory;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.IngestionProperties;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.result.OperationStatus;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.source.CompressionType;
import com.microsoft.Azure Synapse Analytics.kusto.ingest.source.StreamSourceInfo;
import java.io.FileInputStream;
import java.io.InputStream;

public class FileIngestion {
    public static void main(String[] args) throws Exception {
        String poolPath = "https://<Poolname>.<WorkspaceName>.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(
                poolPath,
                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 conjunto de Data Explorer

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 conjunto de 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 conjunto de 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 Azure Synapse Studio ou o portal do Azure.

Studio

1. No Synapse Studio, no painel esquerdo, selecione Desenvolver.

2. Em Scripts KQL, selecione + (Adicionar novo recurso) >script KQL. No painel do lado direito, pode atribuir um nome ao script.

3. No menu Ligar a , selecione contosodataexplorer.

4. No menu Utilizar base de dados , selecione TestDatabase.

5. Cole o seguinte comando e selecione Executar para criar a tabela.

   .delete table TestTable policy streamingingestion
   

6. Na portal do Azure, aceda ao conjunto de Data Explorer.

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

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

9. Selecione Guardar.

Portal

1. Na portal do Azure, aceda ao conjunto de 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 Desativado para desativar a Ingestão de transmissão em fluxo.

5. Selecione Guardar.

Limitações

• Os cursores de base de dados não são suportados para uma base de dados se a própria base de dados ou qualquer uma das respetivas tabelas tiver a política de ingestão de transmissão em fluxo definida e ativada.
• 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 recentemente ingeridos 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 for utilizada em qualquer uma das tabelas da base de dados, esta base de dados não pode ser utilizada como líder para bases de dados de seguidores ou como fornecedor de dados para Azure Synapse Analytics Data Share.

Passos seguintes

• Analisar com Data Explorer
• Monitorizar conjuntos de Data Explorer