Configurare l'inserimento in streaming nel pool Esplora dati di Azure Synapse (Anteprima)

L’inserimento in streaming è utile per il caricamento dei dati quando è necessaria una bassa latenza tra inserimento e query. Si consideri l'uso di un inserimento in streaming negli scenari seguenti:

• È obbligatoria una latenza inferiore a un secondo.
• Per ottimizzare l'elaborazione operativa di molte tabelle in cui il flusso di dati in ogni tabella è relativamente piccolo (pochi record al secondo), ma il volume di inserimento dati complessivo è elevato (migliaia di record al secondo).

Se il flusso di dati in ogni tabella è elevato (oltre 4 GB all'ora), è consigliabile usare l'inserimento batch.

Per altre informazioni sui diversi metodi di inserimento, vedere Panoramica dell'inserimento dati.

Scegliere il tipo di inserimento in streaming appropriato

Sono supportati due tipi di inserimento in streaming:

Tipo di inserimento	Descrizione
Hub eventi o Hub IoT	Gli hub vengono configurati come origini dati di streaming delle tabelle. Per informazioni sulla configurazione di queste impostazioni, vedere Hub eventi.
Inserimento personalizzato	Per l'inserimento personalizzato è necessario scrivere un'applicazione che usa una delle librerie client di Esplora dati di Azure Synapse. Usare le informazioni contenute in questo argomento per configurare l'inserimento personalizzato. Anche l'applicazione di esempio di inserimento in streaming C# può risultare utile.

Usare la tabella seguente per scegliere il tipo di inserimento appropriato per l'ambiente di utilizzo:

Criterio	Hub eventi/Hub IoT	Inserimento personalizzato
Ritardo dei dati tra l'avvio dell'inserimento e la disponibilità dei dati per la query	Ritardo più lungo	Ritardo più breve
Sovraccarico di sviluppo	Configurazione rapida e semplice, senza sovraccarico di sviluppo	Sovraccarico elevato di sviluppo per creare un'applicazione per inserire i dati, gestire gli errori e garantire la coerenza dei dati

Nota

L'inserimento di dati da un hub eventi nei pool Esplora dati non sarà possibile se l'area di lavoro di Synapse usa una rete virtuale gestita con protezione dall'esfiltrazione di dati abilitata.

Prerequisiti

• Una sottoscrizione di Azure. Creare un account Azure gratuito.

• Creare un pool Esplora dati usando Synapse Studio o il portale di Azure

• Creare un database di Esplora dati.

  1. Nel riquadro sinistro di Synapse Studio selezionare Dati.

  2. Selezionare + (Aggiungi nuova risorsa) >Pool Esplora dati e usare le informazioni seguenti:

     Impostazione	Valore suggerito	Descrizione
     Nome pool	contosodataexplorer	Nome del pool Esplora dati da usare
     Nome	TestDatabase	Il nome del database deve essere univoco all'interno del cluster.
     Periodo di conservazione predefinito	365	Intervallo di tempo (in giorni) per cui è garantito che i dati rimangano disponibili per le query. L'intervallo di tempo viene misurato dal momento in cui i dati vengono inseriti.
     Periodo cache predefinito	31	L'intervallo di tempo (in giorni) per cui mantenere i dati sottoposti frequentemente a query disponibili nell'archiviazione su unità SSD o nella RAM, invece che nell'archiviazione a lungo termine.

  3. Selezionare Crea per creare il database. Per la creazione è in genere necessario meno di un minuto.

• Ottenere gli endpoint di query e inserimento dati.

  1. Nel riquadro sinistro di Synapse Studio selezionare Gestisci>Pool Esplora dati.

  2. Selezionare il pool Esplora dati da usare per visualizzarne i dettagli.

     

  3. Prendere nota degli endpoint di query e inserimento dati. Usare l'endpoint Query come cluster durante la configurazione delle connessioni al pool Esplora dati. Quando si configurano gli SDK per l'inserimento dati, usare l'endpoint di inserimento dati.

     

Considerazioni sulle prestazioni e sulle operazioni

I collaboratori principali che possono influire sull'inserimento in streaming sono:

• Specifica di calcolo: le prestazioni e la capacità di inserimento in streaming aumentano le dimensioni del pool Esplora dati. Il numero di richieste di inserimento simultanee è limitato a sei per core. Ad esempio, per un tipo di carico di lavoro a 16 core, come Ottimizzato per il calcolo (Grande) e Ottimizzato per l’archiviazione (Grande), il carico massimo supportato è di 96 richieste di inserimento simultanee. Per due tipi di carico di lavoro core, ad esempio Ottimizzato per il calcolo (Extra Small), il carico massimo supportato è di 12 richieste di inserimento simultanee.
• Limite di dimensioni dei dati: il limite di dimensioni dei dati per una richiesta di inserimento in streaming è di 4 MB.
• Aggiornamenti dello schema: gli aggiornamenti dello schema, ad esempio la creazione e la modifica di tabelle e mapping di inserimento, possono impiegare fino a cinque minuti per il servizio di inserimento in streaming. Per altre informazioni, vedere Inserimento in streaming e modifiche allo schema.
• Capacità SSD: abilitando l'inserimento in streaming in un pool Esplora dati, persino quando i dati non vengono inseriti tramite streaming, parte del disco SSD locale dei computer del pool Esplora dati viene usata per lo streaming dei dati di inserimento e lo spazio di archiviazione disponibile per la cache ad accesso frequente viene ridotto.

Abilitare l'inserimento in streaming nel pool Esplora dati

Prima di poter usare l'inserimento in streaming, è necessario abilitare la funzionalità nel pool Esplora dati e definire criteri di inserimento in streaming. È possibile abilitare la funzionalità durante la creazione del pool Esplora dati o aggiungerla a un pool Esplora dati esistente.

Avviso

Esaminare le limitazioni prima di abilitare l'inserimento in streaming.

Abilitare l'inserimento in streaming durante la creazione di un nuovo pool Esplora dati

È possibile abilitare l'inserimento in streaming durante la creazione di un nuovo pool Esplora dati usando Azure Synapse Studio o il portale di Azure.

Studio

Durante la creazione di un pool Esplora dati usando la procedura descritta in Creare un pool Esplora dati con Synapse Studio, nella scheda Impostazioni aggiuntive selezionare Inserimento in streaming> Abilitato.

Portale

Durante la creazione di un pool Esplora dati usando la procedura descritta in Creare un pool Esplora dati usando il portale di Azure, nella scheda Impostazioni aggiuntive selezionare Inserimento in streaming> Abilitato.

Abilitare l'inserimento in streaming in un pool Esplora dati esistente

Se si dispone di un pool Esplora dati esistente, è possibile abilitare l'inserimento in streaming tramite il portale di Azure.

1. Nel portale di Azure andare al pool Esplora dati.
2. In Impostazioniselezionare Configurazioni.
3. Nel riquadro Configurazioni selezionare On per abilitare l'inserimento in streaming.
4. Seleziona Salva.

Creare una tabella di destinazione e definire i criteri

Creare una tabella per ricevere i dati di inserimento in streaming e definirne i criteri correlati usando Azure Synapse Studio o il portale di Azure.

Studio

1. Nel riquadro sinistro di Synapse Studio selezionare Sviluppo.

2. In Script KQL selezionare + (Aggiungi nuova risorsa) >Script KQL. Nel riquadro a destra è possibile assegnare un nome allo script.

3. Nel menu Connetti a selezionare contosodataexplorer.

4. Nel menu Usa database selezionare TestDatabase.

5. Incollare il comando seguente e selezionare Esegui per creare una tabella.

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

6. Copiare uno dei comandi seguenti nel riquadro Query e selezionare Esegui. In questo modo vengono definiti i criteri di inserimento in streaming nella tabella creata o nel database che contiene la tabella.

   Suggerimento

   Un criterio definito a livello di database si applica a tutte le tabelle esistenti e future presenti nel database.

   • Per definire i criteri nella tabella creata, usare:

     .alter table TestTable policy streamingingestion enable
     

   • Per definire i criteri nel database contenente la tabella creata, usare:

     .alter database StreamingTestDb policy streamingingestion enable
     

Portale

1. Nel portale di Azure Synapse Analytics andare al pool Esplora dati.

2. Seleziona Query.

3. Per creare la tabella che riceverà i dati tramite l'inserimento in streaming, copiare il comando seguente nel riquadro Query e selezionare Esegui.

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

4. Copiare uno dei comandi seguenti nel riquadro Query e selezionare Esegui. In questo modo vengono definiti i criteri di inserimento in streaming nella tabella creata o nel database che contiene la tabella.

   Suggerimento

   Un criterio definito a livello di database si applica a tutte le tabelle esistenti e future presenti nel database.

   • Per definire i criteri nella tabella creata, usare:

     .alter table TestTable policy streamingingestion enable
     

   • Per definire i criteri nel database contenente la tabella creata, usare:

     .alter database StreamingTestDb policy streamingingestion enable
     

Creare un'applicazione di inserimento in streaming per inserire dati nel pool Esplora dati

Creare l'applicazione per l'inserimento di dati nel pool Esplora dati usando il linguaggio preferito. Per la variabile poolPath, usare l'endpoint Query annotato in Prerequisiti.

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

Go

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

Disabilitare l'inserimento in streaming nel pool Esplora dati

Avviso

La disabilitazione dell'inserimento in streaming può richiedere alcune ore.

Prima di disabilitare l'inserimento in streaming nel pool Esplora dati, eliminare i criteri di inserimento in streaming da tutte le tabelle e i database pertinenti. La rimozione dei criteri di inserimento in streaming attiva la ridistribuzione dei dati all'interno del pool Esplora dati. I dati di inserimento in streaming sono spostati dalla risorsa di archiviazione iniziale a quella di archiviazione permanente nell'archivio colonne (extent o partizioni). Questo processo può richiedere da alcuni secondi a poche ore, a seconda della quantità di dati presenti nella risorsa di archiviazione iniziale.

Rimuovere i criteri di inserimento in streaming

È possibile eliminare i criteri di inserimento in streaming usando Azure Synapse Studio o il portale di Azure.

Studio

1. Nel riquadro sinistro di Synapse Studio selezionare Sviluppo.

2. In Script KQL selezionare + (Aggiungi nuova risorsa) >Script KQL. Nel riquadro a destra è possibile assegnare un nome allo script.

3. Nel menu Connetti a selezionare contosodataexplorer.

4. Nel menu Usa database selezionare TestDatabase.

5. Incollare il comando seguente e selezionare Esegui per creare una tabella.

   .delete table TestTable policy streamingingestion
   

6. Nel portale di Azure andare al pool Esplora dati.

7. In Impostazioniselezionare Configurazioni.

8. Nel riquadro Configurazioni selezionare On per abilitare l'inserimento in streaming.

9. Seleziona Salva.

Portale

1. Nel portale di Azure andare al pool Esplora dati e selezionare Query.

2. Per eliminare i criteri di inserimento in streaming dalla tabella, copiare il comando seguente nel riquadro Query e selezionare Esegui.

   .delete table TestTable policy streamingingestion
   

3. In Impostazioniselezionare Configurazioni.

4. Nel riquadro Configurazioni selezionare Off per disabilitare l'inserimento in streaming.

5. Seleziona Salva.

Limiti

• I cursori di database non sono supportati per un database se il database stesso o una delle relative tabelle dispone dei criteri di inserimento in streaming definiti e abilitati.
• I mapping dei dati devono essere pre-creati per l'uso nell'inserimento in streaming. Le singole richieste di inserimento in streaming non supportano i mapping dei dati inline.
• Non è possibile impostare i tag extent nei dati di inserimento in streaming.
• Aggiorna criteri. I criteri di aggiornamento possono fare riferimento solo ai dati appena inseriti nella tabella di origine e non ad altri dati o tabelle nel database.
• Se l'inserimento in streaming viene usato in una delle tabelle del database, non è possibile usare quest’ultimo come leader per i database follower o come provider di dati per la condivisione di dati di Azure Synapse Analytics.

Passaggi successivi

• Analizzare con Esplora dati
• Monitorare i pool Esplora dati