Konfigurera strömmande inmatning på din Azure Synapse Data Explorer-pool (förhandsversion)

Direktuppspelningsinmatning är användbart för att läsa in data när du behöver korta svarstider mellan inmatning och fråga. Överväg att använda strömningsinmatning i följande scenarier:

• Svarstid på mindre än en sekund krävs.
• Optimera driftbearbetningen av många tabeller där dataströmmen till varje tabell är relativt liten (några poster per sekund), men den totala datainmatningsvolymen är hög (tusentals poster per sekund).

Om dataströmmen i varje tabell är hög (över 4 GB per timme) bör du överväga att använda batchinmatning.

Mer information om olika inmatningsmetoder finns i översikten över datainmatning.

Välj lämplig typ av direktuppspelningsinmatning

Två typer av direktuppspelningsinmatning stöds:

Inmatningstyp	Beskrivning
Händelsehubb eller IoT Hub	Hubbar konfigureras som datakällor för tabellströmning. Information om hur du konfigurerar dessa finns i Event Hub.
Anpassad inmatning	Anpassad inmatning kräver att du skriver ett program som använder något av de Azure Synapse Data Explorer klientbiblioteken. Använd informationen i det här avsnittet för att konfigurera anpassad inmatning. Du kanske också tycker att C#-exempelprogrammet för direktuppspelning är användbart.

Använd följande tabell för att hjälpa dig att välja den inmatningstyp som är lämplig för din miljö:

Kriterium	Händelsehubb/IoT Hub	Anpassad inmatning
Datafördröjning mellan inmatningsinitiering och tillgängliga data för fråga	Längre fördröjning	Kortare fördröjning
Utvecklingskostnader	Snabb och enkel installation, inga utvecklingskostnader	Höga utvecklingskostnader för att skapa ett program som matar in data, hanterar fel och säkerställer datakonsekvens

Anteckning

Att mata in data från en händelsehubb i Data Explorer pooler fungerar inte om synapse-arbetsytan använder ett hanterat virtuellt nätverk med dataexfiltreringsskydd aktiverat.

Förutsättningar

• En Azure-prenumeration. Skapa ett kostnadsfritt Azure-konto.

• Skapa en Data Explorer pool med Synapse Studio eller Azure Portal

• Skapa en Data Explorer databas.

  1. I Synapse Studio väljer du Data i fönstret till vänster.

  2. Välj + (Lägg till ny resurs) >Data Explorer pool och använd följande information:

     Inställning	Föreslaget värde	Beskrivning
     Poolnamn	contosodataexplorer	Namnet på den Data Explorer pool som ska användas
     Name	TestDatabase	Databasnamnet måste vara unikt inom klustret.
     Standardkvarhållningsperiod	365	Det tidsintervall (i dagar) då det är garanterat att data förblir tillgängliga för frågor. Tidsintervallet mäts från det att data matas in.
     Standardcacheperiod	31	Det tidsintervall (i dagar) då data som frågor körs mot ofta ska vara tillgängliga i SSD-lagring eller RAM i stället för i långsiktig lagring.

  3. Välj Skapa för att skapa databasen. Det brukar ta mindre än en minut att skapa en databas.

• Hämta slutpunkterna fråga och datainmatning.

  1. I Synapse Studio väljer du Hantera>Data Explorer pooler i fönstret till vänster.

  2. Välj den Data Explorer pool som du vill använda för att visa dess information.

     

  3. Anteckna fråge- och datainmatningsslutpunkterna. Använd frågeslutpunkten som kluster när du konfigurerar anslutningar till din Data Explorer pool. När du konfigurerar SDK:er för datainmatning använder du slutpunkten för datainmatning.

     

Prestanda- och driftsöverväganden

De främsta bidragsgivarna som kan påverka strömmande inmatning är:

• Beräkningsspecifikation: Strömmande inmatningsprestanda och kapacitetsskalor med ökade Data Explorer poolstorlekar. Antalet samtidiga inmatningsbegäranden är begränsat till sex per kärna. För arbetsbelastningstypen 16 kärnor, till exempel Beräkningsoptimerad (stor) och Lagringsoptimerad (stor), är den maximala belastningen 96 samtidiga inmatningsbegäranden. För två kärnarbetsbelastningstyper, till exempel Beräkningsoptimerad (extra liten), är den maximala belastningen som stöds 12 samtidiga inmatningsbegäranden.
• Datastorleksgräns: Datastorleksgränsen för en begäran om strömmande inmatning är 4 MB.
• Schemauppdateringar: Schemauppdateringar, till exempel skapande och ändring av tabeller och inmatningsmappningar, kan ta upp till fem minuter för tjänsten för direktuppspelning. Mer information finns i Strömmande inmatning och schemaändringar.
• SSD-kapacitet: Aktivera strömmande inmatning på en Data Explorer pool, även om data inte matas in via strömning, använder en del av den lokala SSD-disken på Data Explorer pooldatorer för strömmande inmatningsdata och minskar lagringen som är tillgänglig för frekvent cache.

Aktivera strömningsinmatning i din Data Explorer pool

Innan du kan använda strömningsinmatning måste du aktivera funktionen i din Data Explorer-pool och definiera en princip för inmatning av direktuppspelning. Du kan aktivera funktionen när du skapar Data Explorer-poolen eller lägga till den i en befintlig Data Explorer pool.

Varning

Granska begränsningarna innan du aktiverar strömningsinmatning.

Aktivera strömningsinmatning när du skapar en ny Data Explorer pool

Du kan aktivera direktuppspelningsinmatning när du skapar en ny Data Explorer-pool med Azure Synapse Studio eller Azure Portal.

Studio

När du skapar en Data Explorer pool med hjälp av stegen i Skapa en Data Explorer pool med Synapse Studio går du till fliken Ytterligare inställningar och väljer Direktuppspelningsinmatning>aktiverad.

Portal

När du skapar en Data Explorer pool med hjälp av stegen i Skapa en Data Explorer pool med hjälp av Azure Portal väljer du Strömmande inmatning>aktiverad på fliken Ytterligare inställningar.

Aktivera strömningsinmatning i en befintlig Data Explorer pool

Om du har en befintlig Data Explorer pool kan du aktivera strömningsinmatning med hjälp av Azure Portal.

1. I Azure Portal går du till din Data Explorer pool.
2. I Inställningar väljer du Konfigurationer.
3. I fönstret Konfigurationer väljer du På för att aktivera strömningsinmatning.
4. Välj Spara.

Skapa en måltabell och definiera principen

Skapa en tabell för att ta emot strömmande inmatningsdata och definiera dess relaterade princip med hjälp av Azure Synapse Studio eller Azure Portal.

Studio

1. I Synapse Studio väljer du Utveckla i fönstret till vänster.

2. Under KQL-skript väljer du + (Lägg till ny resurs) >KQL-skript. I den högra rutan kan du namnge skriptet.

3. På menyn Anslut till väljer du contosodataexplorer.

4. På menyn Använd databas väljer du TestDatabase.

5. Klistra in följande kommando och välj Kör för att skapa tabellen.

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

6. Kopiera något av följande kommandon till fönstret Fråga och välj Kör. Detta definierar principen för strömningsinmatning i den tabell som du skapade eller i databasen som innehåller tabellen.

   Tips

   En princip som definieras på databasnivå gäller för alla befintliga och framtida tabeller i databasen.

   • Om du vill definiera principen i tabellen som du skapade använder du:

     .alter table TestTable policy streamingingestion enable
     

   • Om du vill definiera principen för databasen som innehåller tabellen du skapade använder du:

     .alter database StreamingTestDb policy streamingingestion enable
     

Portal

1. I Azure Synapse Analytics-portalen går du till din Data Explorer-pool.

2. Välj Fråga.

3. Om du vill skapa tabellen som ska ta emot data via strömmande inmatning kopierar du följande kommando till fönstret Fråga och väljer Kör.

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

4. Kopiera något av följande kommandon till fönstret Fråga och välj Kör. Detta definierar principen för strömningsinmatning i den tabell som du skapade eller i databasen som innehåller tabellen.

   Tips

   En princip som definieras på databasnivå gäller för alla befintliga och framtida tabeller i databasen.

   • Om du vill definiera principen i tabellen som du skapade använder du:

     .alter table TestTable policy streamingingestion enable
     

   • Om du vill definiera principen för databasen som innehåller tabellen du skapade använder du:

     .alter database StreamingTestDb policy streamingingestion enable
     

Skapa ett program för direktuppspelningsinmatning för att mata in data till din Data Explorer pool

Skapa ditt program för att mata in data till din Data Explorer pool med det språk du föredrar. För variabeln poolPath använder du den frågeslutpunkt som du antecknar i Krav.

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

Kör

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

Inaktivera strömningsinmatning i din Data Explorer-pool

Varning

Det kan ta några timmar att inaktivera strömningsinmatning.

Innan du inaktiverar strömningsinmatning på din Data Explorer-pool släpper du principen för strömmande inmatning från alla relevanta tabeller och databaser. Borttagningen av den strömmande inmatningsprincipen utlöser omorganisering av data i din Data Explorer-pool. Strömmande inmatningsdata flyttas från den ursprungliga lagringen till permanent lagring i kolumnlagret (utrymmen eller shards). Den här processen kan ta mellan några sekunder och några timmar, beroende på mängden data i den ursprungliga lagringen.

Ta bort principen för strömningsinmatning

Du kan ta bort principen för strömningsinmatning med hjälp av Azure Synapse Studio eller Azure Portal.

Studio

1. I Synapse Studio väljer du Utveckla i fönstret till vänster.

2. Under KQL-skript väljer du + (Lägg till ny resurs) >KQL-skript. I den högra rutan kan du namnge skriptet.

3. På menyn Anslut till väljer du contosodataexplorer.

4. På menyn Använd databas väljer du TestDatabase.

5. Klistra in följande kommando och välj Kör för att skapa tabellen.

   .delete table TestTable policy streamingingestion
   

6. I Azure Portal går du till din Data Explorer pool.

7. I Inställningar väljer du Konfigurationer.

8. I fönstret Konfigurationer väljer du På för att aktivera strömningsinmatning.

9. Välj Spara.

Portal

1. I Azure Portal går du till din Data Explorer-pool och väljer Fråga.

2. Om du vill ta bort den strömmande inmatningsprincipen från tabellen kopierar du följande kommando till frågefönstret och väljer Kör.

   .delete table TestTable policy streamingingestion
   

3. I Inställningar väljer du Konfigurationer.

4. I fönstret Konfigurationer väljer du Av för att inaktivera strömningsinmatning.

5. Välj Spara.

Begränsningar

• Databasmarkörer stöds inte för en databas om själva databasen eller någon av dess tabeller har principen För direktuppspelning definierad och aktiverad.
• Datamappningar måste skapas i förväg för användning vid strömningsinmatning. Enskilda begäranden om direktuppspelningsinmatning har inte plats för infogade datamappningar.
• Det går inte att ange utrymmestaggar för strömmande inmatningsdata.
• Uppdatera princip. Uppdateringsprincipen kan endast referera till nyligen inmatade data i källtabellen och inte andra data eller tabeller i databasen.
• Om strömningsinmatning används i någon av databastabellerna kan den här databasen inte användas som ledare för uppföljningsdatabaser eller som dataprovider för Azure Synapse Analytics-Data Share.

Nästa steg

• Analysera med Data Explorer
• Övervaka Data Explorer pooler