Konfigurieren der Streamingerfassung in Ihrem Azure Synapse Data Explorer-Pool (Vorschau)

Artikel
06/01/2023

Die Streamingerfassung ist nützlich zum Laden von Daten, wenn eine geringe Wartezeit zwischen Erfassung und Abfrage erforderlich ist. Die Streamingerfassung kann in folgenden Szenarien verwendet werden:

Eine Wartezeit von weniger als einer Sekunde ist erforderlich.
Optimierung der operativen Verarbeitung vieler Tabellen, bei denen der Datenstrom für die einzelnen Tabellen jeweils relativ klein (wenige Datensätze pro Sekunde), das Gesamtvolumen der Datenerfassung aber hoch ist (mehrere tausend Datensätze pro Sekunde).

Wenn der Datenstrom für die einzelnen Tabellen groß ist (über 4 GB pro Stunde), sollten Sie ggf. die Batcherfassung nutzen.

Unter Übersicht über die Datenerfassung in Azure Data Explorer erfahren Sie mehr über verschiedene Erfassungsmethoden.

Auswählen des geeigneten Streamingerfassungstyps

Zwei Streamingerfassungstypen werden unterstützt:

Erfassungstyp	BESCHREIBUNG
Event Hub oder IoT Hub	Hubs werden als Datenquellen für das Tabellenstreaming konfiguriert. Informationen zu ihrer Einrichtung finden Sie unter Event Hub.
Benutzerdefinierte Erfassung	Für die benutzerdefinierte Erfassung muss eine Anwendung geschrieben werden, die eine der Clientbibliotheken von Azure Synapse Data Explorer verwendet. Konfigurieren Sie die benutzerdefinierte Erfassung anhand der Informationen in diesem Thema. Unter Umständen finden Sie auch die C#-Beispielanwendung für die Streamingerfassung hilfreich.

Die folgende Tabelle unterstützt Sie beim Auswählen eines für Ihre Umgebung geeigneten Erfassungstyps:

Kriterium	Event Hub/IoT Hub	Benutzerdefinierte Erfassung
Datenverzögerung zwischen der Initiierung der Erfassung und der Verfügbarkeit der Daten für Abfragen	Längere Verzögerung	Kürzere Verzögerung
Zusätzlicher Entwicklungsaufwand	Schnelle und einfache Einrichtung, kein zusätzlicher Entwicklungsaufwand	Hoher zusätzlicher Entwicklungsaufwand zum Erstellen einer Anwendung, um die Daten zu erfassen, Fehler zu behandeln und die Datenkonsistenz sicherzustellen.

Hinweis

Das Erfassen von Daten aus einem Event Hub in Data Explorer-Pools funktioniert nicht, wenn Ihr Synapse-Arbeitsbereich ein verwaltetes virtuelles Netzwerk mit aktiviertem Datenexfiltrationsschutz verwendet.

Voraussetzungen

Ein Azure-Abonnement. Erstellen Sie ein kostenloses Azure-Konto.
Erstellen eines Data Explorer-Pools über Synapse Studio oder im Azure-Portal

Erstellen Sie eine Data Explorer-Datenbank.

Wählen Sie in Synapse Studio im linken Bereich Daten aus.

Wählen Sie + (Neue Ressource hinzufügen) >Data Explorer-Pool aus, und verwenden Sie die folgenden Informationen:

Einstellung	Vorgeschlagener Wert	BESCHREIBUNG
Poolname	contosodataexplorer	Name des zu verwendende Data Explorer-Pools
Name	TestDatabase	Der Datenbankname muss innerhalb des Clusters eindeutig sein.
Standardaufbewahrungszeitraum	365	Die Zeitspanne (in Tagen), für die garantiert wird, dass die Daten für Abfragen verfügbar bleiben. Die Zeitspanne wird ab dem Zeitpunkt gemessen, zu dem die Daten erfasst werden.
Standardcachezeitraum	31	Die Zeitspanne (in Tagen), wie lange häufig abgefragte Daten im SSD-Speicher oder RAM (und nicht im längerfristigen Speicher) verfügbar bleiben.

Wählen Sie Erstellen, um die Datenbank zu erstellen. Die Erstellung dauert in der Regel weniger als eine Minute.

Abrufen der Abfrage- und Datenerfassungsendpunkte
1. Wählen Sie in Synapse Studio im linken Bereich die Option Verwalten>Data Explorer-Pools aus.
2. Wählen Sie den Data Explorer-Pool aus, zu dem Sie Details anzeigen möchten.
3. Notieren Sie sich die Abfrage- und Datenerfassungsendpunkte. Verwenden Sie den Abfrageendpunkt als Cluster, wenn Sie Verbindungen mit Ihrem Data Explorer-Pool konfigurieren. Verwenden Sie beim Konfigurieren von SDKs für die Datenerfassung den Datenerfassungsendpunkt.

Überlegungen zu Leistung und Betrieb

Die wichtigsten Aspekte, die sich auf die Streamingerfassung auswirken können:

Computespezifikation: Die Leistung und Kapazität der Streamingerfassung wird bei größeren Data Explorer-Pools skaliert. Die Anzahl der gleichzeitigen Erfassungsanforderungen ist auf sechs pro Kern beschränkt. Für 16 Kernworkloadtypen wie „Für Compute optimiert“ (Groß) und „Datenspeicheroptimiert“ (Groß) beträgt die maximal unterstützte Last beispielsweise 96 gleichzeitige Erfassungsanforderungen. Für zwei Kernworkloadtypen wie „Für Compute optimiert“ (Sehr klein) beträgt die maximal unterstützte Last beispielsweise 12 gleichzeitige Erfassungsanforderungen.
Datengrößenbeschränkung: Die Datengröße für Anforderungen zur Streamingerfassung ist auf 4 MB beschränkt.
Schemaaktualisierungen: Schemaaktualisierungen wie die Erstellung und Änderung von Tabellen und Erfassungszuordnungen können für den Streamingerfassungsdienst bis zu fünf Minuten dauern. Weitere Informationen finden Sie unter Streamingerfassung und Schemaänderungen.
SSD-Kapazität: Wenn die Streamingerfassung in einem Data Explorer-Pool aktiviert wird, wird ein Teil des lokalen SSD-Datenträgers der Data Explorer-Poolcompuer für Streamingerfassungsdaten genutzt, wodurch sich der verfügbare Speicherplatz für den aktiven Cache verringert. Dies gilt auch, wenn gar keine Daten per Streaming erfasst werden.

Aktivieren der Streamingerfassung für Ihren Data Explorer-Pool

Bevor Sie die Streamingerfassung verwenden können, müssen Sie die Funktion für den Data Explorer-Pool aktivieren und eine Streamingerfassungsrichtlinie definieren. Sie können die Funktion beim Erstellen des Data Explorer-Pools aktivieren oder einem vorhandenen Data Explorer-Pool hinzufügen.

Warnung

Überprüfen Sie vor dem Aktivieren der Streamingerfassung die Einschränkungen.

Aktivieren der Streamingerfassung beim Erstellen eines neuen Data Explorer-Pools

Sie können die Streamingerfassung beim Erstellen eines neuen Data Explorer-Pools mithilfe von Azure Synapse Studio oder des Azure-Portals aktivieren.

Studio
Portal

Wählen Sie beim Erstellen eines Data Explorer-Pools anhand der Schritte unter Schnellstart: Erstellen eines Data Explorer-Pools mit Synapse Studio (Vorschau) auf der Registerkarte Zusätzliche Einstellungen die Optionen Streamingerfassung>Aktiviert aus.

Aktivieren der Streamingerfassung beim Erstellen eines Data Explorer-Pools in Azure Synapse Data Explorer

Aktivieren der Streamingerfassung für einen vorhandenen Data Explorer-Pool

Wenn Sie über einen vorhandenen Data Explorer-Pool verfügen, können Sie die Streamingerfassung mithilfe des Azure-Portals aktivieren.

Navigieren Sie im Azure-Portal zu Ihrem Data Explorer-Pool.
Wählen Sie unter Einstellungen die Option Konfigurationen aus.
Wählen Sie im Bereich Konfigurationen die Option Ein aus, um die Streamingerfassung zu aktivieren.
Wählen Sie Speichern aus.

Erstellen einer Zieltabelle und Definieren der Richtlinie

Erstellen Sie eine Tabelle, um die Streamingerfassungsdaten zu empfangen, und definieren Sie die zugehörige Richtlinie mithilfe von Azure Synapse Studio oder des Azure-Portals.

Studio
Portal

Wählen Sie in Synapse Studio im linken Bereich Entwickeln aus.
Wählen Sie unter KQL-Skripts die Option + (Neue Ressource hinzufügen) >KQL-Skript aus. Im rechten Bereich können Sie Ihr Skript benennen.
Wählen Sie im Menü Verbinden mit den Eintrag contosodataexplorer aus.
Wählen Sie im Menü Datenbank verwenden die Option TestDatabase aus.
Fügen Sie den folgenden Befehl ein, und wählen Sie Ausführen aus, um die Tabelle zu erstellen.
```
.create table TestTable (TimeStamp: datetime, Name: string, Metric: int, Source:string)
```
Kopieren Sie einen der folgenden Befehle in den Abfragebereich, und wählen Sie Ausführen aus: Dadurch wird die Streamingerfassungsrichtlinie für die von Ihnen erstellte Tabelle oder für die Datenbank definiert, die diese Tabelle enthält.

Tipp

Eine Richtlinie, die auf Datenbankebene definiert ist, gilt für alle vorhandenen und zukünftigen Tabellen in der Datenbank.
- Verwenden Sie Folgendes, um die Richtlinie für die von Ihnen erstellte Tabelle zu definieren:
```
.alter table TestTable policy streamingingestion enable
```
- Verwenden Sie Folgendes, um die Richtlinie für die Datenbank zu definieren, die die von Ihnen erstellte Tabelle enthält:
```
.alter database StreamingTestDb policy streamingingestion enable
```

Navigieren Sie im Azure Synapse Analytics-Portal zu Ihrem Data Explorer-Pool.
Wählen Sie Abfrage.
Kopieren Sie den folgenden Befehl in den Abfragebereich, und wählen Sie Ausführen aus, um die Tabelle zu erstellen, die die Daten über Streamingerfassung erhalten soll:
```
.create table TestTable (TimeStamp: datetime, Name: string, Metric: int, Source:string)
```
Kopieren Sie einen der folgenden Befehle in den Abfragebereich, und wählen Sie Ausführen aus: Dadurch wird die Streamingerfassungsrichtlinie für die von Ihnen erstellte Tabelle oder für die Datenbank definiert, die diese Tabelle enthält.

Tipp

Eine Richtlinie, die auf Datenbankebene definiert ist, gilt für alle vorhandenen und zukünftigen Tabellen in der Datenbank.
- Verwenden Sie Folgendes, um die Richtlinie für die von Ihnen erstellte Tabelle zu definieren:
```
.alter table TestTable policy streamingingestion enable
```
- Verwenden Sie Folgendes, um die Richtlinie für die Datenbank zu definieren, die die von Ihnen erstellte Tabelle enthält:
```
.alter database StreamingTestDb policy streamingingestion enable
```

Erstellen einer Anwendung für die Streamingerfassung zum Erfassen von Daten in Ihrem Data Explorer-Pool

Erstellen Sie Ihre Anwendung zum Erfassen von Daten in Ihrem Data Explorer-Pool unter Verwendung der von Ihnen bevorzugten Sprache. Verwenden Sie für die Variable poolPath den Abfrageendpunkt, den Sie sich unter Voraussetzungen notiert haben.

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

Deaktivieren der Streamingerfassung für Ihren Data Explorer-Pool

Warnung

Die Deaktivierung der Streamingerfassung kann mehrere Stunden dauern.

Bevor Sie die Streamingerfassung für Ihren Data Explorer-Pool deaktivieren, löschen Sie die Streamingerfassungsrichtlinie aus allen relevanten Tabellen und Datenbanken. Durch das Entfernen der Streamingerfassungsrichtlinie wird eine Neuanordnung der Daten in Ihrem Data Explorer-Pool ausgelöst. Die Streamingerfassungsdaten werden aus dem anfänglichen Speicher in den permanenten Speicher im Spaltenspeicher (Erweiterungen oder Shards) verschoben. Dieser Vorgang kann abhängig von der Datenmenge im anfänglichen Speicher zwischen einigen Sekunden und wenigen Stunden dauern.

Löschen der Streamingerfassungsrichtlinie

Sie können die Streamingerfassungsrichtlinie mithilfe von Azure Synapse Studio oder des Azure-Portals löschen.

Studio
Portal

Wählen Sie in Synapse Studio im linken Bereich Entwickeln aus.
Wählen Sie unter KQL-Skripts die Option + (Neue Ressource hinzufügen) >KQL-Skript aus. Im rechten Bereich können Sie Ihr Skript benennen.
Wählen Sie im Menü Verbinden mit den Eintrag contosodataexplorer aus.
Wählen Sie im Menü Datenbank verwenden die Option TestDatabase aus.
Fügen Sie den folgenden Befehl ein, und wählen Sie Ausführen aus, um die Tabelle zu erstellen.
```
.delete table TestTable policy streamingingestion
```
Navigieren Sie im Azure-Portal zu Ihrem Data Explorer-Pool.
Wählen Sie unter Einstellungen die Option Konfigurationen aus.
Wählen Sie im Bereich Konfigurationen die Option Ein aus, um die Streamingerfassung zu aktivieren.
Wählen Sie Speichern aus.

Navigieren Sie im Azure-Portal zu Ihrem Data Explorer-Pool, und wählen Sie Abfragen aus.
Kopieren Sie zum Löschen der Streamingerfassungsrichtlinie aus der Tabelle den folgenden Befehl in den Abfragebereich, und wählen Sie Ausführen aus:
```
.delete table TestTable policy streamingingestion
```
Wählen Sie unter Einstellungen die Option Konfigurationen aus.
Wählen Sie im Bereich Konfigurationen die Option Off aus, um die Streamingerfassung zu aktivieren.
Wählen Sie Speichern aus.

Einschränkungen

Datenbankcursor werden für eine Datenbank nicht unterstützt, wenn für diese selbst oder für eine Ihrer Tabellen die Richtlinie für die Streamingerfassung definiert und aktiviert wurde.
Datenzuordnungen müssen vorab erstellt werden, damit sie bei der Streamingerfassung verwendet werden können. Individuelle Anforderungen zur Streamingerfassung bieten keine Inlinedatenzuordnungen.
Erweiterungstags können für die Streamingerfassungsdaten nicht festgelegt werden.
Aktualisierungsrichtlinie: Die Aktualisierungsrichtlinie kann nur auf die neu erfassten Daten in der Quelltabelle und nicht auf andere Daten oder Tabellen in der Datenbank verweisen.
Wenn die Streamingerfassung für eine der Tabellen der Datenbank verwendet wird, kann diese Datenbank nicht als Leader für Follower-Datenbanken oder als Datenanbieter für die Azure Synapse Analytics-Datenfreigabe verwendet werden.