Configurer l’ingestion en streaming dans votre pool Azure Synapse Data Explorer (préversion)

L’ingestion en streaming permet de charger des données lorsque vous avez besoin d’une faible latence entre l’ingestion et l’interrogation. Vous pouvez utiliser l’ingestion en streaming dans les scénarios suivants :

• Lorsqu’une latence inférieure à une seconde est nécessaire.
• Pour optimiser le traitement opérationnel d’un grand nombre de tables, lorsque le flux de données de chaque table est relativement faible (quelques enregistrements par seconde), mais que le volume global d’ingestion de données est élevé (des milliers d’enregistrements par seconde).

Si le flux de données de chaque table est élevé (plus de 4 Go par heure), nous vous conseillons d’utiliser l’ingestion par lots.

Pour plus d’informations sur les méthodes d’ingestion, consultez Vue d’ensemble de l’ingestion des données.

Choisir le type d’ingestion de streaming approprié

Deux types d’ingestion de streaming sont pris en charge :

Type d’ingestion	Description
Event Hub ou IoT Hub	Les hubs sont configurés comme des sources de données de streaming de table. Pour plus d’informations sur la configuration de ces paramètres, consultez Event Hub.
Ingestion personnalisée	Une ingestion personnalisée vous demande d’écrire une application qui utilise l’une des bibliothèques de client Azure Synapse Data Explorer. Utilisez les informations de cette rubrique pour configurer l’ingestion personnalisée. L’exemple d’application pour l’ingestion en streaming avec C# peut également vous être utile.

Aidez-vous du tableau suivant pour choisir le type d’ingestion le mieux adapté à votre environnement :

Critère	Event Hub/IoT Hub	Ingestion personnalisée
Délai de données entre le lancement de l’ingestion et le moment où les données sont disponibles pour une requête	Délai plus long	Délai plus court
Surcharge de développement	Installation rapide et facile, aucune surcharge de développement	Frais de développement élevés pour créer une application, ingérer des données, gérer les erreurs et garantir la cohérence des données

Notes

L’ingestion de données à partir d’Event Hub dans des pools de l’Explorateur de données ne fonctionne pas si votre espace de travail Synapse utilise un réseau virtuel managé avec la protection contre l’exfiltration des données activée.

Prérequis

• Un abonnement Azure. Créez un compte Azure gratuit.

• Créez un pool Data Explorer en utilisant Synapse Studio ou le portail Azure

• Créez une base de données Data Explorer.

  1. Dans Synapse Studio, dans le volet de gauche, sélectionnez Données.

  2. Sélectionnez +(Ajouter une nouvelle ressource) >Pool Data Explorer et utilisez les informations suivantes :

     Paramètre	Valeur suggérée	Description
     Nom du pool	contosodataexplorer	Nom du pool Data Explorer à utiliser
     Nom	TestDatabase	Ce nom de base de données doit être unique dans le cluster.
     Période de conservation par défaut	365	Intervalle de temps (en jours) pendant lequel vous avez la garantie d’avoir les données à disposition pour les interroger. Cet intervalle se mesure à partir du moment où les données sont ingérées.
     Période de cache par défaut	31	Intervalle de temps (en jours) pendant lequel les données fréquemment interrogées restent disponibles dans le stockage SSD ou la RAM, plutôt que dans un stockage à plus long terme.

  3. Sélectionnez Créer pour créer la base de données. La création prend généralement moins d’une minute.

• Obtenez les points de terminaison de requête et d’ingestion de données.

  1. Dans Synapse Studio, dans le volet de gauche, sélectionnez Gérer>Pools Data Explorer.

  2. Sélectionnez le pool Data Explorer à utiliser pour voir ses détails.

     

  3. Notez les points de terminaison de requête et d’ingestion de données. Utilisez le point de terminaison de requête comme cluster pour la configuration des connexions à votre pool Data Explorer. Lors de la configuration des kits de développement logiciel (SDK) pour l’ingestion des données, utilisez le point de terminaison d’ingestion des données.

     

Considérations relatives aux performances et aux opérations

Les contributeurs principaux qui peuvent impacter l’ingestion en streaming sont les suivants :

• Spécification de calcul : niveau de performance de l’ingestion en streaming et échelles de capacité avec des tailles de pool Data Explorer accrues. Le nombre de demandes d’ingestion simultanées est limité à six par cœur. Par exemple, pour un type de charge de travail à 16 cœurs, tel qu’Optimisé pour le calcul (Grand) et À stockage optimisé (Grand), la charge maximale prise en charge est de 96 demandes d’ingestion simultanées. Pour 2 types de charges de travail de base, tels qu’Optimisé pour le calcul (Très petit), la charge maximale prise en charge est de 12 demandes d’ingestion simultanées.
• Limite de taille des données : la taille limite des données par demande d’ingestion en streaming est de 4 Mo.
• Mises à jour des schémas : les mises à jour des schémas, telles que la création et la modification des tables et des mappages d’ingestion, peuvent prendre jusqu’à cinq minutes pour le service d’ingestion de streaming. Pour plus d’informations, consultez Ingestion de streaming et changements de schéma.
• Capacité du disque SSD : l’activation de l’ingestion en streaming sur un pool Data Explorer, même lorsque les données ne sont pas ingérées via streaming, utilise une partie du disque SSD local des machines du pool Data Explorer pour les données d’ingestion en streaming et réduit le stockage disponible pour le cache chaud.

Activer l’ingestion en streaming sur votre pool Data Explorer

Avant de pouvoir utiliser l’ingestion en streaming, vous devez activer cette capacité sur votre pool Data Explorer et définir une stratégie d’ingestion en streaming. Vous pouvez activer cette capacité lors de la création du pool Data Explorer ou l’ajouter à un pool Data Explorer existant.

Avertissement

Avant d’activer l’ingestion en streaming, passez en revue les limitations.

Activer l’ingestion en streaming lors de la création d’un nouveau pool Data Explorer

Vous pouvez activer l’ingestion en streaming lors de la création d’un pool Data Explorer à l’aide d’Azure Synapse Studio ou du portail Azure.

Studio

Lorsque vous créez un pool Data Explorer à l’aide des étapes décrites dans Créer un pool Data Explorer à l’aide de Synapse Studio, dans l’onglet Paramètres supplémentaires, sélectionnez Ingestion en streaming>Activée.

Portail

Lorsque vous créez un pool Data Explorer à l’aide des étapes décrites dans Créer un pool Data Explorer à l’aide du portail Azure, dans l’onglet Paramètres supplémentaires, sélectionnez Ingestion en streaming>Activée.

Activer l’ingestion en streaming sur un pool Data Explorer existant

Si vous disposez d’un pool Data Explorer existant, vous pouvez activer l’ingestion en streaming à l’aide du portail Azure.

1. Dans le portail Azure, accédez à votre pool Data Explorer.
2. Dans Paramètres, sélectionnez Configurations.
3. Dans le volet Configurations, sélectionnez Activé pour activer Ingestion de streaming.
4. Sélectionnez Enregistrer.

Créer une table cible et définir la stratégie

Créez une table pour recevoir les données d’ingestion en streaming et définissez la stratégie associée à l’aide d’Azure Synapse Studio ou du portail Azure.

Studio

1. Dans Synapse Studio, dans le volet de gauche, sélectionnez Développer.

2. Sous Scripts KQL, sélectionnez + (Ajouter une nouvelle ressource) >Script KQL. Dans le volet de droite, vous pouvez nommer votre script.

3. Dans le menu Connecter à, sélectionnez contosodataexplorer.

4. Dans le menu Utiliser la base de données, sélectionnez TestDatabase.

5. Collez la commande suivante, puis sélectionnez Exécuter pour créer la table.

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

6. Copiez l’une des commandes suivantes dans le volet de requête, puis sélectionnez Exécuter. Cela définit la stratégie d’ingestion en streaming pour la table que vous avez créée ou la base de données qui contient cette table.

   Conseil

   Une stratégie qui est définie au niveau de la base de données s’applique à toutes les tables existantes et futures de la base de données.

   • Pour définir la stratégie de la table que vous avez créée, utilisez ceci :

     .alter table TestTable policy streamingingestion enable
     

   • Pour définir la stratégie de la base de données contenant la table que vous avez créée, utilisez ceci :

     .alter database StreamingTestDb policy streamingingestion enable
     

Portail

1. Dans le portail Azure Synapse Analytics, accédez à votre pool Data Explorer.

2. Sélectionnez Requête.

3. Pour créer la table qui recevra les données par le biais de l’ingestion de streaming, copiez la commande suivante dans le volet de requête, puis sélectionnez Exécuter.

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

4. Copiez l’une des commandes suivantes dans le volet de requête, puis sélectionnez Exécuter. Cela définit la stratégie d’ingestion en streaming pour la table que vous avez créée ou la base de données qui contient cette table.

   Conseil

   Une stratégie qui est définie au niveau de la base de données s’applique à toutes les tables existantes et futures de la base de données.

   • Pour définir la stratégie de la table que vous avez créée, utilisez ceci :

     .alter table TestTable policy streamingingestion enable
     

   • Pour définir la stratégie de la base de données contenant la table que vous avez créée, utilisez ceci :

     .alter database StreamingTestDb policy streamingingestion enable
     

Créer une application d’ingestion en streaming pour ingérer des données dans votre pool Data Explorer

Créez votre application pour ingérer des données dans votre pool Data Explorer à l’aide de votre langage préféré. Pour la variable poolPath, utilisez le point de terminaison de requête que vous avez enregistré dans les prérequis.

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

Désactiver l’ingestion en streaming sur votre pool Data Explorer

Avertissement

La désactivation de l’ingestion de streaming peut prendre plusieurs heures.

Avant de désactiver l’ingestion en streaming sur votre pool Data Explorer, supprimez la stratégie d’ingestion en streaming de toutes les tables et bases de données concernées. La suppression de la stratégie d’ingestion en streaming déclenche la réorganisation des données au sein de votre pool Data Explorer. Les données d’ingestion de streaming sont déplacées du stockage initial vers le stockage permanent dans la banque de colonnes (étendues ou partitions). Ce processus peut prendre de quelques secondes à quelques heures, selon la quantité de données qui se trouvent dans le stockage initial.

Supprimer la stratégie d’ingestion en streaming

Vous pouvez supprimer la stratégie d’ingestion en streaming à l’aide d’Azure Synapse Studio ou du portail Azure.

Studio

1. Dans Synapse Studio, dans le volet de gauche, sélectionnez Développer.

2. Sous Scripts KQL, sélectionnez + (Ajouter une nouvelle ressource) >Script KQL. Dans le volet de droite, vous pouvez nommer votre script.

3. Dans le menu Connecter à, sélectionnez contosodataexplorer.

4. Dans le menu Utiliser la base de données, sélectionnez TestDatabase.

5. Collez la commande suivante, puis sélectionnez Exécuter pour créer la table.

   .delete table TestTable policy streamingingestion
   

6. Dans le portail Azure, accédez à votre pool Data Explorer.

7. Dans Paramètres, sélectionnez Configurations.

8. Dans le volet Configurations, sélectionnez Activé pour activer Ingestion de streaming.

9. Sélectionnez Enregistrer.

Portail

1. Dans le portail Azure, accédez à votre pool Data Explorer et sélectionnez Requête.

2. Pour supprimer la stratégie d’ingestion de streaming de la table, copiez la commande suivante dans le volet de requête, puis sélectionnez Exécuter.

   .delete table TestTable policy streamingingestion
   

3. Dans Paramètres, sélectionnez Configurations.

4. Dans le volet Configurations, sélectionnez Désactivé pour désactiver Ingestion de streaming.

5. Sélectionnez Enregistrer.

Limites

• Les curseurs de base de données ne sont pas pris en charge pour une base de données si une stratégie d’ingestion de streaming est définie et activée pour la base de données ou l’une de ses tables.
• Les mappages de données doivent être précréés pour être utilisés dans une ingestion de streaming. Les demandes d’ingestion en streaming individuelles ne prennent pas en charge les mappages de données inline.
• Les étiquettes d’étendue ne peuvent pas être définies dans les données d’ingestion de streaming.
• Stratégie de mise à jour. La stratégie de mise à jour ne peut référencer que les données nouvellement ingérées dans la table source et aucune autre donnée ou table de la base de données.
• Si l’ingestion en streaming est utilisée sur l’une des tables de la base de données, cette dernière ne peut pas être utilisée en tant que leader pour les bases de données abonnées, ni en tant que fournisseur de données pour Azure Synapse Analytics Data Share.

Étapes suivantes

• Analyser avec l’Explorateur de données
• Surveiller les pools de l’Explorateur de données