Konfigurowanie pozyskiwania przesyłania strumieniowego w puli Azure Synapse Data Explorer (wersja zapoznawcza)

Pozyskiwanie przesyłania strumieniowego jest przydatne do ładowania danych, gdy potrzebujesz małego opóźnienia między pozyskiwaniem i wykonywaniem zapytań. Rozważ użycie pozyskiwania przesyłania strumieniowego w następujących scenariuszach:

• Wymagane jest opóźnienie mniejsze niż sekundę.
• Aby zoptymalizować przetwarzanie operacyjne wielu tabel, w których strumień danych do każdej tabeli jest stosunkowo mały (kilka rekordów na sekundę), ale ogólny wolumin pozyskiwania danych jest wysoki (tysiące rekordów na sekundę).

Jeśli strumień danych do każdej tabeli jest wysoki (ponad 4 GB na godzinę), rozważ użycie pozyskiwania wsadowego.

Aby dowiedzieć się więcej o różnych metodach pozyskiwania danych, zobacz Omówienie pozyskiwania danych.

Wybierz odpowiedni typ pozyskiwania przesyłania strumieniowego

Obsługiwane są dwa typy pozyskiwania przesyłania strumieniowego:

Typ pozyskiwania	Opis
Centrum zdarzeń lub IoT Hub	Koncentratory są konfigurowane jako źródła danych przesyłania strumieniowego tabel. Aby uzyskać informacje na temat konfigurowania tych ustawień, zobacz Centrum zdarzeń.
Niestandardowe pozyskiwanie	Niestandardowe pozyskiwanie wymaga zapisania aplikacji korzystającej z jednej z bibliotek klienta Azure Synapse Data Explorer. Skorzystaj z informacji w tym temacie, aby skonfigurować niestandardowe pozyskiwanie. Przykładowa aplikacja pozyskiwania przesyłania strumieniowego w języku C# może być również przydatna.

Poniższa tabela ułatwia wybór typu pozyskiwania, który jest odpowiedni dla danego środowiska:

Kryterium	Centrum zdarzeń / IoT Hub	Niestandardowe pozyskiwanie
Opóźnienie danych między inicjowaniem pozyskiwania a danymi dostępnymi dla zapytania	Dłuższe opóźnienie	Krótsze opóźnienie
Nakład pracy nad programowaniem	Szybka i łatwa konfiguracja, bez obciążeń programistycznych	Wysokie obciążenie programistyczne w celu utworzenia aplikacji pozyskiwania danych, obsługi błędów i zapewnienia spójności danych

Uwaga

Pozyskiwanie danych z centrum zdarzeń do pul Data Explorer nie będzie działać, jeśli obszar roboczy usługi Synapse używa zarządzanej sieci wirtualnej z włączoną ochroną eksfiltracji danych.

Wymagania wstępne

• Subskrypcja platformy Azure. Utwórz bezpłatne konto platformy Azure.

• Tworzenie puli Data Explorer przy użyciu Synapse Studio lub Azure Portal

• Utwórz bazę danych Data Explorer.

  1. W Synapse Studio w okienku po lewej stronie wybierz pozycję Dane.

  2. Wybierz + pozycję (Dodaj nowy zasób) >Data Explorer puli i użyj następujących informacji:

     Ustawienie	Sugerowana wartość	Opis
     Nazwa puli	contosodataexplorer	Nazwa puli Data Explorer do użycia
     Nazwa	TestDatabase	Nazwa bazy danych musi być unikatowa w obrębie klastra.
     Domyślny okres przechowywania	365	Przedział czasu (w dniach), w którym gwarantowana jest dostępność danych dla zapytania. Przedział czasu jest mierzony od momentu pozyskania danych.
     Domyślny okres pamięci podręcznej	31	Przedział czasu (w dniach), w którym często używane w zapytaniach dane mają być dostępne na dysku SSD lub w pamięci RAM zamiast w magazynie długoterminowym.

  3. Wybierz pozycję Utwórz, aby utworzyć bazę danych. Tworzenie zazwyczaj zajmuje mniej niż minutę.

• Pobierz punkty końcowe pozyskiwania zapytań i danych.

  1. W Synapse Studio w okienku po lewej stronie wybierz pozycję Zarządzaj>Data Explorer pule.

  2. Wybierz pulę Data Explorer, której chcesz użyć, aby wyświetlić jego szczegóły.

     

  3. Zanotuj punkty końcowe pozyskiwania zapytań i danych. Użyj punktu końcowego zapytania jako klastra podczas konfigurowania połączeń z pulą Data Explorer. Podczas konfigurowania zestawów SDK na potrzeby pozyskiwania danych użyj punktu końcowego pozyskiwania danych.

     

Zagadnienia dotyczące wydajności i działania

Głównymi współautorami, którzy mogą mieć wpływ na pozyskiwanie przesyłania strumieniowego, są:

• Specyfikacja obliczeniowa: wydajność pozyskiwania przesyłania strumieniowego i skalowanie pojemności dzięki zwiększonym rozmiarom puli Data Explorer. Liczba współbieżnych żądań pozyskiwania jest ograniczona do sześciu na rdzeń. Na przykład w przypadku 16-rdzeniowego typu obciążenia, takiego jak Zoptymalizowane pod kątem obliczeń (duże) i Zoptymalizowane pod kątem magazynu (duże), maksymalne obsługiwane obciążenie wynosi 96 współbieżnych żądań pozyskiwania. W przypadku dwóch podstawowych typów obciążeń, takich jak Compute Optimized (Extra Small), maksymalne obsługiwane obciążenie to 12 współbieżnych żądań pozyskiwania.
• Limit rozmiaru danych: limit rozmiaru danych dla żądania pozyskiwania przesyłania strumieniowego wynosi 4 MB.
• Aktualizacje schematu: aktualizacje schematu, takie jak tworzenie i modyfikowanie tabel i mapowań pozyskiwania, może potrwać do pięciu minut w przypadku usługi pozyskiwania przesyłania strumieniowego. Aby uzyskać więcej informacji, zobacz Przesyłanie strumieniowe pozyskiwania i zmiany schematu.
• Pojemność ssd: włączenie pozyskiwania przesyłania strumieniowego w puli Data Explorer, nawet jeśli dane nie są pozyskiwane za pośrednictwem przesyłania strumieniowego, używa części lokalnego dysku SSD maszyn puli Data Explorer do przesyłania strumieniowego danych pozyskiwania i zmniejsza ilość miejsca dostępnego dla gorącej pamięci podręcznej.

Włączanie pozyskiwania przesyłania strumieniowego w puli Data Explorer

Aby można było korzystać z pozyskiwania przesyłania strumieniowego, należy włączyć funkcję w puli Data Explorer i zdefiniować zasady pozyskiwania przesyłania strumieniowego. Możliwość można włączyć podczas tworzenia puli Data Explorer lub dodać ją do istniejącej puli Data Explorer.

Ostrzeżenie

Zapoznaj się z ograniczeniami przed włączeniem pozyskiwania przesyłania strumieniowego.

Włączanie pozyskiwania przesyłania strumieniowego podczas tworzenia nowej puli Data Explorer

Pozyskiwanie przesyłania strumieniowego można włączyć podczas tworzenia nowej puli Data Explorer przy użyciu programu Azure Synapse Studio lub Azure Portal.

Studio

Podczas tworzenia puli Data Explorer, wykonując kroki opisane w temacie Tworzenie puli Data Explorer przy użyciu Synapse Studio, na karcie Dodatkowe ustawienia wybierz pozycjęWłączone pozyskiwanie >przesyłania strumieniowego.

Portal

Podczas tworzenia puli Data Explorer, wykonując kroki opisane w temacie Tworzenie puli Data Explorer przy użyciu Azure Portal, na karcie Dodatkowe ustawienia wybierz pozycjęWłączone pozyskiwanie >przesyłania strumieniowego.

Włączanie pozyskiwania przesyłania strumieniowego w istniejącej puli Data Explorer

Jeśli masz istniejącą pulę Data Explorer, możesz włączyć pozyskiwanie przesyłania strumieniowego przy użyciu Azure Portal.

1. W Azure Portal przejdź do puli Data Explorer.
2. W obszarze Ustawienia wybierz pozycję Konfiguracje.
3. W okienku Konfiguracje wybierz pozycję Włączone , aby włączyć pozyskiwanie przesyłania strumieniowego.
4. Wybierz pozycję Zapisz.

Tworzenie tabeli docelowej i definiowanie zasad

Utwórz tabelę, aby odbierać dane pozyskiwania przesyłania strumieniowego i definiować powiązane zasady przy użyciu programu Azure Synapse Studio lub Azure Portal.

Studio

1. W Synapse Studio w okienku po lewej stronie wybierz pozycję Programowanie.

2. W obszarze Skrypty KQL wybierz + (Dodaj nowy zasób) >skrypt KQL. W okienku po prawej stronie możesz nazwać skrypt.

3. W menu Połącz z wybierz pozycję contosodataexplorer.

4. W menu Użyj bazy danych wybierz pozycję TestDatabase.

5. Wklej następujące polecenie i wybierz pozycję Uruchom , aby utworzyć tabelę.

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

6. Skopiuj jedno z następujących poleceń do okienka Zapytanie i wybierz pozycję Uruchom. Definiuje zasady pozyskiwania danych przesyłanych strumieniowo w tabeli utworzonej lub w bazie danych zawierającej tabelę.

   Porada

   Zasady zdefiniowane na poziomie bazy danych mają zastosowanie do wszystkich istniejących i przyszłych tabel w bazie danych.

   • Aby zdefiniować zasady w utworzonej tabeli, użyj:

     .alter table TestTable policy streamingingestion enable
     

   • Aby zdefiniować zasady w bazie danych zawierającej utworzoną tabelę, użyj:

     .alter database StreamingTestDb policy streamingingestion enable
     

Portal

1. W portalu usługi Azure Synapse Analytics przejdź do puli Data Explorer.

2. Wybierz pozycję Zapytanie.

3. Aby utworzyć tabelę, która będzie odbierać dane za pośrednictwem pozyskiwania danych przesyłanych strumieniowo, skopiuj następujące polecenie do okienka Zapytanie i wybierz pozycję Uruchom.

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

4. Skopiuj jedno z następujących poleceń do okienka Zapytanie i wybierz pozycję Uruchom. Definiuje zasady pozyskiwania danych przesyłanych strumieniowo w tabeli utworzonej lub w bazie danych zawierającej tabelę.

   Porada

   Zasady zdefiniowane na poziomie bazy danych mają zastosowanie do wszystkich istniejących i przyszłych tabel w bazie danych.

   • Aby zdefiniować zasady w utworzonej tabeli, użyj:

     .alter table TestTable policy streamingingestion enable
     

   • Aby zdefiniować zasady w bazie danych zawierającej utworzoną tabelę, użyj:

     .alter database StreamingTestDb policy streamingingestion enable
     

Tworzenie aplikacji pozyskiwania strumieniowego w celu pozyskiwania danych do puli Data Explorer

Utwórz aplikację na potrzeby pozyskiwania danych do puli Data Explorer przy użyciu preferowanego języka. Dla zmiennej poolPath użyj punktu końcowego zapytania zanotowego w sekcji Wymagania wstępne.

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

Przejdź

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

Wyłączanie pozyskiwania danych przesyłanych strumieniowo w puli Data Explorer

Ostrzeżenie

Wyłączenie pozyskiwania danych przesyłanych strumieniowo może potrwać kilka godzin.

Przed wyłączeniem pozyskiwania danych przesyłanych strumieniowo w puli Data Explorer usuń zasady pozyskiwania przesyłania strumieniowego ze wszystkich odpowiednich tabel i baz danych. Usunięcie zasad pozyskiwania przesyłania strumieniowego powoduje ponowne rozmieszczenie danych w puli Data Explorer. Dane pozyskiwania danych przesyłanych strumieniowo są przenoszone z początkowego magazynu do magazynu trwałego w magazynie kolumn (zakresy lub fragmenty). Ten proces może potrwać od kilku sekund do kilku godzin, w zależności od ilości danych w początkowym magazynie.

Usuwanie zasad pozyskiwania przesyłania strumieniowego

Zasady pozyskiwania przesyłania strumieniowego można usunąć przy użyciu programu Azure Synapse Studio lub Azure Portal.

Studio

1. W Synapse Studio w okienku po lewej stronie wybierz pozycję Opracuj.

2. W obszarze Skrypty KQL wybierz + skrypt KQL (Dodaj nowy zasób). > W okienku po prawej stronie możesz nazwać skrypt.

3. W menu Połącz z wybierz pozycję contosodataexplorer.

4. W menu Użyj bazy danych wybierz pozycję TestDatabase.

5. Wklej następujące polecenie i wybierz pozycję Uruchom , aby utworzyć tabelę.

   .delete table TestTable policy streamingingestion
   

6. W Azure Portal przejdź do puli Data Explorer.

7. W obszarze Ustawienia wybierz pozycję Konfiguracje.

8. W okienku Konfiguracje wybierz pozycję Włączone , aby włączyć pozyskiwanie danych przesyłanych strumieniowo.

9. Wybierz pozycję Zapisz.

Portal

1. W Azure Portal przejdź do puli Data Explorer i wybierz pozycję Zapytanie.

2. Aby usunąć zasady pozyskiwania danych przesyłanych strumieniowo z tabeli, skopiuj następujące polecenie do okienka Zapytanie i wybierz pozycję Uruchom.

   .delete table TestTable policy streamingingestion
   

3. W obszarze Ustawienia wybierz pozycję Konfiguracje.

4. W okienku Konfiguracje wybierz pozycję Wyłączone , aby wyłączyć pozyskiwanie przesyłania strumieniowego.

5. Wybierz pozycję Zapisz.

Ograniczenia

• Kursory bazy danych nie są obsługiwane dla bazy danych, jeśli sama baza danych lub którakolwiek z jej tabel mają zdefiniowane i włączone zasady pozyskiwania przesyłania strumieniowego .
• Mapowania danych muszą być wstępnie utworzone do użycia w pozyskiwaniu danych przesyłanych strumieniowo. Pojedyncze żądania pozyskiwania danych przesyłanych strumieniowo nie są uwzględniane w wbudowanych mapowaniach danych.
• Nie można ustawić tagów zakresu na danych pozyskiwania strumieniowego.
• Zaktualizuj zasady. Zasady aktualizacji mogą odwoływać się tylko do nowo pozyskanych danych w tabeli źródłowej, a nie do żadnych innych danych lub tabel w bazie danych.
• Jeśli pozyskiwanie strumieniowe jest używane w żadnej z tabel bazy danych, tej bazy danych nie można użyć jako lidera dla baz danych obserwowanych ani jako dostawcy danych dla Azure Synapse Analytics Data Share.

Następne kroki

• Analizowanie za pomocą Data Explorer
• Monitorowanie pul Data Explorer