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

Lors de la création d’un pool Data Explorer à l’aide des étapes de création d’un pool Data Explorer à l’aide de Synapse Studio, sous l’onglet Paramètres supplémentaires, sélectionnez Ingestion> de streamingactivée.

Lors de la création d’un pool Data Explorer à l’aide des étapes de création d’un pool Data Explorer à l’aide du portail Azure, sous l’onglet Paramètres supplémentaires, sélectionnez Ingestion> de streamingactivée.

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 Requête et sélectionnez Exécuter. Cela définit la stratégie d’ingestion de streaming sur la table que vous avez créée ou sur la base de données qui contient la table.

   Conseil / Astuce

   Une stratégie 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 sur la table que vous avez créée, utilisez :

     .alter table TestTable policy streamingingestion enable
     

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

     .alter database StreamingTestDb policy streamingingestion enable
     

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 via l’ingestion de streaming, copiez la commande suivante dans le volet Requête et 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 Requête et sélectionnez Exécuter. Cela définit la stratégie d’ingestion de streaming sur la table que vous avez créée ou sur la base de données qui contient la table.

   Conseil / Astuce

   Une stratégie 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 sur la table que vous avez créée, utilisez :

     .alter table TestTable policy streamingingestion enable
     

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

     .alter database StreamingTestDb policy streamingingestion enable
     

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

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) 

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

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

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

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 l’ingestion de streaming.

9. Cliquez sur Enregistrer.

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 diffusion en continu à partir de la table, copiez la commande suivante dans le volet Requête et 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 l’ingestion de streaming.

5. Cliquez sur Enregistrer.