Streamingopname configureren in uw Azure Synapse Data Explorer-pool (preview)

Streamingopname is handig voor het laden van gegevens wanneer u een lage latentie tussen opname en query nodig hebt. Overweeg het gebruik van streamingopname in de volgende scenario's:

• Latentie van minder dan een seconde is vereist.
• De operationele verwerking van veel tabellen optimaliseren waarbij de gegevensstroom in elke tabel relatief klein is (een paar records per seconde), maar het totale gegevensopnamevolume hoog is (duizenden records per seconde).

Als de gegevensstroom in elke tabel hoog is (meer dan 4 GB per uur), kunt u overwegen batchopname te gebruiken.

Zie Overzicht van gegevensopname voor meer informatie over verschillende opnamemethoden.

Kies het juiste type streamingopname

Er worden twee typen streamingopname ondersteund:

Opnametype	Description
Event Hub of IoT Hub	Hubs worden geconfigureerd als gegevensbronnen voor tabelstreaming. Zie Event Hub voor meer informatie over het instellen hiervan.
Aangepaste opname	Voor aangepaste opname moet u een toepassing schrijven die gebruikmaakt van een van de Azure Synapse Data Explorer clientbibliotheken. Gebruik de informatie in dit onderwerp om aangepaste opname te configureren. Mogelijk vindt u ook de voorbeeldtoepassing voor C#-streamingopname nuttig.

Gebruik de volgende tabel om het opnametype te kiezen dat geschikt is voor uw omgeving:

Criterium	Event Hub/IoT Hub	Aangepaste opname
Gegevensvertraging tussen het starten van de opname en de gegevens die beschikbaar zijn voor query	Langere vertraging	Kortere vertraging
Ontwikkelingsoverhead	Snelle en eenvoudige installatie, geen ontwikkelingsoverhead	Hoge overhead voor ontwikkeling om een toepassing te maken die de gegevens opneemt, fouten afhandelt en gegevensconsistentie garandeert

Notitie

Het opnemen van gegevens uit een Event Hub in Data Explorer-pools werkt niet als uw Synapse-werkruimte gebruikmaakt van een beheerd virtueel netwerk waarvoor bescherming tegen gegevensexfiltratie is ingeschakeld.

Vereisten

• Een Azure-abonnement. Maak een gratis Azure-account.

• Een Data Explorer pool maken met behulp van Synapse Studio of de Azure Portal

• Maak een Data Explorer-database.

  1. Selecteer in Synapse Studio in het linkerdeelvenster De optie Gegevens.

  2. Selecteer + (Nieuwe resource toevoegen) >Data Explorer pool en gebruik de volgende informatie:

     Instelling	Voorgestelde waarde	Beschrijving
     Poolnaam	contosodataexplorer	De naam van de Data Explorer groep die u wilt gebruiken
     Naam	TestDatabase	De databasenaam moet uniek zijn binnen het cluster.
     Standaardretentieperiode	365	De periode (in dagen) dat de gegevens gegarandeerd beschikbaar blijven voor query's. De periode wordt gemeten vanaf het moment dat de gegevens zijn opgenomen.
     Standaardcacheperiode	31	De periode (in dagen) dat vaak opgevraagde gegevens beschikbaar blijven in de SSD-opslag of het RAM-geheugen in plaats van in de langetermijnopslag.

  3. Selecteer Maken om het profiel te maken. Het maakproces duurt meestal minder dan een minuut.

• Haal de eindpunten Query en Gegevensopname op.

  1. Selecteer in Synapse Studio in het linkerdeelvenster Beheren>Data Explorer pools.

  2. Selecteer de Data Explorer pool die u wilt gebruiken om de details ervan weer te geven.

     

  3. Noteer de eindpunten Query en Gegevensopname. Gebruik het Query-eindpunt als het cluster bij het configureren van verbindingen met uw Data Explorer-pool. Gebruik bij het configureren van SDK's voor gegevensopname het eindpunt voor gegevensopname.

     

Prestatie- en operationele overwegingen

De belangrijkste bijdragers die van invloed kunnen zijn op streamingopname zijn:

• Berekeningsspecificatie: prestaties van streamingopname en capaciteit worden geschaald met grotere Data Explorer poolgrootten. Het aantal gelijktijdige opnameaanvragen is beperkt tot zes per kern. Voor een workloadtype met 16 kernen, zoals Geoptimaliseerd voor berekening (groot) en Geoptimaliseerd voor opslag (groot), is de maximaal ondersteunde belasting bijvoorbeeld 96 gelijktijdige opnameaanvragen. Voor twee kernworkloadtypen, zoals Geoptimaliseerd voor rekenkracht (extra klein), is de maximale ondersteunde belasting 12 gelijktijdige opnameaanvragen.
• Limiet voor gegevensgrootte: de limiet voor de gegevensgrootte voor een aanvraag voor streamingopname is 4 MB.
• Schema-updates: Schema-updates, zoals het maken en wijzigen van tabellen en opnametoewijzingen, kunnen maximaal vijf minuten duren voor de service voor streamingopname. Zie Streaming-opname en schemawijzigingen voor meer informatie.
• SSD-capaciteit: het inschakelen van streaming-opname op een Data Explorer-pool, zelfs wanneer gegevens niet via streaming worden opgenomen, gebruikt een deel van de lokale SSD-schijf van de Data Explorer-poolcomputers voor het streamen van opnamegegevens en vermindert de opslag die beschikbaar is voor hot-cache.

Streamingopname inschakelen voor uw Data Explorer-pool

Voordat u streamingopname kunt gebruiken, moet u de mogelijkheid voor uw Data Explorer pool inschakelen en een beleid voor streamingopname definiëren. U kunt de mogelijkheid inschakelen bij het maken van de Data Explorer pool of toevoegen aan een bestaande Data Explorer pool.

Waarschuwing

Bekijk de beperkingen voordat u streamingopname inschakelt.

Streamingopname inschakelen tijdens het maken van een nieuwe Data Explorer-pool

U kunt streamingopname inschakelen tijdens het maken van een nieuwe Data Explorer-pool met behulp van Azure Synapse Studio of de Azure Portal.

Studio

Wanneer u een Data Explorer-pool maakt met behulp van de stappen in Een Data Explorer-pool maken met Synapse Studio, selecteert u op het tabblad Aanvullende instellingen de optie Streamingopname>ingeschakeld.

Portal

Wanneer u een Data Explorer pool maakt met behulp van de stappen in Een Data Explorer pool maken met behulp van de Azure Portal, selecteert u op het tabblad Aanvullende instellingen de optie Streamingopname>ingeschakeld.

Streamingopname inschakelen voor een bestaande Data Explorer-pool

Als u een bestaande Data Explorer-pool hebt, kunt u streamingopname inschakelen met behulp van de Azure Portal.

1. Ga in de Azure Portal naar uw Data Explorer zwembad.
2. Selecteer in Instellingende optie Configuraties.
3. Selecteer in het deelvenster Configuraties de optie Aan om streamingopname in te schakelen.
4. Selecteer Opslaan.

Een doeltabel maken en het beleid definiëren

Maak een tabel om de streamingopnamegegevens te ontvangen en definieer het bijbehorende beleid met behulp van Azure Synapse Studio of de Azure Portal.

Studio

1. Selecteer in Synapse Studio in het linkerdeelvenster Ontwikkelen.

2. Selecteer + onder KQL-scriptsKQL-script (nieuwe resource toevoegen). > In het rechterdeelvenster kunt u het script een naam opgeven.

3. Selecteer in het menu Verbinding maken metde optie contosodataexplorer.

4. Selecteer in het menu Database gebruikende optie TestDatabase.

5. Plak de volgende opdracht en selecteer Uitvoeren om de tabel te maken.

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

6. Kopieer een van de volgende opdrachten naar het deelvenster Query en selecteer Uitvoeren. Hiermee definieert u het streaming-opnamebeleid voor de tabel die u hebt gemaakt of in de database die de tabel bevat.

   Tip

   Een beleid dat is gedefinieerd op databaseniveau, is van toepassing op alle bestaande en toekomstige tabellen in de database.

   • Als u het beleid wilt definiëren voor de tabel die u hebt gemaakt, gebruikt u:

     .alter table TestTable policy streamingingestion enable
     

   • Als u het beleid wilt definiëren voor de database die de tabel bevat die u hebt gemaakt, gebruikt u:

     .alter database StreamingTestDb policy streamingingestion enable
     

Portal

1. Navigeer in de Azure Synapse Analytics-portal naar uw Data Explorer-pool.

2. Selecteer Query.

3. Als u de tabel wilt maken die de gegevens ontvangt via streamingopname, kopieert u de volgende opdracht naar het deelvenster Query en selecteert u Uitvoeren.

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

4. Kopieer een van de volgende opdrachten naar het deelvenster Query en selecteer Uitvoeren. Hiermee definieert u het streaming-opnamebeleid voor de tabel die u hebt gemaakt of in de database die de tabel bevat.

   Tip

   Een beleid dat is gedefinieerd op databaseniveau, is van toepassing op alle bestaande en toekomstige tabellen in de database.

   • Als u het beleid wilt definiëren voor de tabel die u hebt gemaakt, gebruikt u:

     .alter table TestTable policy streamingingestion enable
     

   • Als u het beleid wilt definiëren voor de database die de tabel bevat die u hebt gemaakt, gebruikt u:

     .alter database StreamingTestDb policy streamingingestion enable
     

Een toepassing voor streamingopname maken om gegevens op te nemen in uw Data Explorer-pool

Maak uw toepassing voor het opnemen van gegevens in uw Data Explorer-pool met behulp van uw voorkeurstaal. Gebruik voor de poolPath-variabele het Query-eindpunt dat u hebt genoteerd in de vereisten.

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

Streamingopname uitschakelen voor uw Data Explorer-pool

Waarschuwing

Het uitschakelen van streamingopname kan enkele uren duren.

Voordat u streamingopname uitschakelt voor uw Data Explorer-pool, moet u het streamingopnamebeleid uit alle relevante tabellen en databases verwijderen. Als het streaming-opnamebeleid wordt verwijderd, worden gegevens opnieuw in uw Data Explorer-pool gegroepeerd. De streamingopnamegegevens worden verplaatst van de initiële opslag naar de permanente opslag in het kolomarchief (gebieden of shards). Dit proces kan enkele seconden tot een paar uur duren, afhankelijk van de hoeveelheid gegevens in de eerste opslag.

Het streaming-opnamebeleid verwijderen

U kunt het streamingopnamebeleid verwijderen met behulp van Azure Synapse Studio of de Azure Portal.

Studio

1. Selecteer in Synapse Studio in het linkerdeelvenster Ontwikkelen.

2. Selecteer onder KQL-scripts de optie + (Nieuwe resource toevoegen) >KQL-script. In het rechterdeelvenster kunt u het script een naam opgeven.

3. Selecteer in het menu Verbinding maken metde optie contosodataexplorer.

4. Selecteer in het menu Database gebruikende optie TestDatabase.

5. Plak de volgende opdracht en selecteer Uitvoeren om de tabel te maken.

   .delete table TestTable policy streamingingestion
   

6. Ga in de Azure Portal naar uw Data Explorer zwembad.

7. Selecteer in Instellingende optie Configuraties.

8. Selecteer in het deelvenster Configuratiesde optie Aan om streamingopname in te schakelen.

9. Selecteer Opslaan.

Portal

1. Ga in de Azure Portal naar uw Data Explorer-pool en selecteer Query.

2. Als u het streaming-opnamebeleid uit de tabel wilt verwijderen, kopieert u de volgende opdracht naar het deelvenster Query en selecteert u Uitvoeren.

   .delete table TestTable policy streamingingestion
   

3. Selecteer in Instellingende optie Configuraties.

4. Selecteer in het deelvenster Configuratiesde optie Uit om streamingopname uit te schakelen.

5. Selecteer Opslaan.

Beperkingen

• Databasecursors worden niet ondersteund voor een database als voor de database zelf of een van de tabellen het streaming-opnamebeleid is gedefinieerd en ingeschakeld.
• Gegevenstoewijzingen moeten vooraf worden gemaakt voor gebruik bij streamingopname. Afzonderlijke aanvragen voor streamingopname zijn niet geschikt voor inlinegegevenstoewijzingen.
• Bereiktags kunnen niet worden ingesteld voor de streamingopnamegegevens.
• Beleid bijwerken. Het updatebeleid kan alleen verwijzen naar de nieuw opgenomen gegevens in de brontabel en niet naar andere gegevens of tabellen in de database.
• Als streamingopname wordt gebruikt voor een van de tabellen van de database, kan deze database niet worden gebruikt als leider voor volgdatabases of als gegevensprovider voor Azure Synapse Analytics Data Share.

Volgende stappen

• Analyseren met Data Explorer
• Data Explorer-pools bewaken