Streambetöltés konfigurálása a Azure Synapse Data Explorer-készleten (előzetes verzió)

A streambetöltés akkor hasznos az adatok betöltéséhez, ha kis késésre van szüksége a betöltés és a lekérdezés között. Fontolja meg a streambetöltés használatát a következő esetekben:

• Egy másodpercnél rövidebb késésre van szükség.
• Számos olyan tábla működési feldolgozásának optimalizálása érdekében, ahol az adatok adatfolyama az egyes táblákba viszonylag kicsi (másodpercenként néhány rekord), de a teljes adatbetöltési mennyiség magas (több ezer rekord másodpercenként).

Ha az adatok adatfolyama az egyes táblákba magas (óránként 4 GB-nál nagyobb), érdemes lehet kötegelt feldolgozást használni.

A különböző betöltési módszerekkel kapcsolatos további információkért lásd az adatbetöltés áttekintését.

Válassza ki a megfelelő streamelési betöltési típust

Két streamelési betöltési típus támogatott:

Betöltési típus	Description
Event Hub vagy IoT Hub	A hubok táblastreamelési adatforrásként vannak konfigurálva. A beállítással kapcsolatos információkért lásd: Event Hub.
Egyéni betöltés	Az egyéni betöltéshez olyan alkalmazást kell írnia, amely a Azure Synapse Data Explorer ügyfélkódtárak egyikét használja. Az ebben a témakörben található információk segítségével konfigurálhatja az egyéni betöltési adatokat. A C#-streamelési mintaalkalmazás is hasznos lehet.

Az alábbi táblázat segítségével kiválaszthatja a környezetének megfelelő betöltési típust:

Kritérium	Event Hub /IoT Hub	Egyéni betöltés
Adatbetöltés kezdeményezése és a lekérdezéshez elérhető adatok közötti adatkésés	Hosszabb késleltetés	Rövidebb késleltetés
Fejlesztési többletterhelés	Gyors és egyszerű beállítás, nincs fejlesztési többletterhelés	Nagy fejlesztési terhelés az adatok betöltését, a hibák kezelését és az adatkonzisztenciát biztosító alkalmazások létrehozásához

Megjegyzés

Az eseményközpontból Data Explorer készletekbe való adatbetöltés nem fog működni, ha a Synapse-munkaterület felügyelt virtuális hálózatot használ, amelyen engedélyezve van az adatkiszivárgás elleni védelem.

Előfeltételek

• Azure-előfizetés. Hozzon létre egy ingyenes Azure-fiókot.

• Data Explorer-készlet létrehozása Synapse Studio vagy a Azure Portal használatával

• Hozzon létre egy Data Explorer adatbázist.

  1. A Synapse Studio bal oldali ablaktábláján válassza az Adatok lehetőséget.

  2. Válassza az + (Új erőforrás hozzáadása) >Data Explorer készletet, és használja a következő információkat:

     Beállítás	Ajánlott érték	Leírás
     Készlet neve	contosodataexplorer	A használni kívánt Data Explorer készlet neve
     Name	TestDatabase	Az adatbázis nevének egyedinek kell lennie a fürtön belül.
     Alapértelmezett megőrzési idő	365	Az az időtartomány (napokban), amelyre vonatkozóan garantált, hogy az adatok elérhetők maradnak a lekérdezéshez. Az időtartam az adatok betöltésének időpontjával kezdődik.
     Alapértelmezett gyorsítótár-időszak	31	Az az időtartomány (napokban), amelynél a gyakran lekérdezett adatok nem hosszabb távú, hanem SSD-tárolóban vagy RAM-ban érhetők el.

  3. Az adatbázis létrehozásához válassza a Létrehozás lehetőséget. Az adatbázis létrehozása általában kevesebb mint egy percet vesz igénybe.

• Kérje le a lekérdezési és adatbetöltési végpontokat.

  1. A Synapse Studio bal oldali ablaktábláján válassza a Kezelés>Data Explorer készletek lehetőséget.

  2. Válassza ki azt a Data Explorer készletet, amelyet a részletek megtekintéséhez használni szeretne.

     

  3. Jegyezze fel a lekérdezési és adatbetöltési végpontokat. A lekérdezési végpontot használja fürtként a Data Explorer készlethez való kapcsolatok konfigurálásakor. Az SDK-k adatbetöltésre való konfigurálásakor használja az adatbetöltési végpontot.

     

Teljesítménnyel és működésével kapcsolatos szempontok

A streambetöltést befolyásoló fő közreműködők a következők:

• Számítási specifikáció: A streambetöltési teljesítmény és a kapacitás nagyobb Data Explorer készletmérettel skálázható. Az egyidejű betöltési kérelmek száma magonként legfeljebb hat lehet. Például 16 magos számítási feladattípus esetén, például a Számításoptimalizált (Nagy) és a Tárhelyoptimalizált (Nagy) esetén a maximálisan támogatott terhelés 96 egyidejű betöltési kérés. Két alapvető számítási feladattípus, például a Számításoptimalizált (Extra kicsi) esetében a maximálisan támogatott terhelés 12 egyidejű betöltési kérelem.
• Adatméretkorlát: A streamelési betöltési kérelem adatméretkorlátja 4 MB.
• Sémafrissítések: A sémafrissítések, például a táblák és a betöltési leképezések létrehozása és módosítása akár öt percet is igénybe vehet a streamelési szolgáltatás számára. További információ: Streambetöltés és sémamódosítások.
• SSD-kapacitás: A streambetöltés engedélyezése Data Explorer készleten, még akkor is, ha az adatok nem streameléssel vannak betöltve, a Data Explorer készlet gépeinek helyi SSD-lemezének egy részét használja a betöltési adatok streamelésére, és csökkenti a gyorsgyorsítótár számára elérhető tárhelyet.

Streambetöltés engedélyezése a Data Explorer-készleten

Mielőtt használhatja a streambetöltést, engedélyeznie kell a képességet a Data Explorer készleten, és meg kell határoznia egy streamelési betöltési szabályzatot. Engedélyezheti a képességet a Data Explorer készlet létrehozásakor, vagy hozzáadhatja egy meglévő Data Explorer készlethez.

Figyelmeztetés

A streambetöltés engedélyezése előtt tekintse át a korlátozásokat .

Streambetöltés engedélyezése új Data Explorer készlet létrehozásakor

A streambetöltést engedélyezheti új Data Explorer-készlet létrehozásakor a Azure Synapse Studio vagy a Azure Portal használatával.

Studio

Amikor Data Explorer készletet hoz létre a Data Explorer-készlet létrehozása Synapse Studio használatával című cikk lépéseit követve, a További beállítások lapon válassza a Streambetöltés>engedélyezve lehetőséget.

Portál

Amikor Data Explorer készletet hoz létre a Data Explorer készlet létrehozása a Azure Portal használatával című cikk lépéseit követve, a További beállítások lapon válassza a Streambetöltés>engedélyezve lehetőséget.

Streambetöltés engedélyezése meglévő Data Explorer készleten

Ha már van Data Explorer készlete, engedélyezheti a streambetöltést a Azure Portal használatával.

1. A Azure Portal lépjen a Data Explorer készletre.
2. A Beállítások területen válassza a Konfigurációk lehetőséget.
3. A Konfigurációk panelen válassza a Be lehetőséget a streambetöltés engedélyezéséhez.
4. Válassza a Mentés lehetőséget.

Céltábla létrehozása és a szabályzat meghatározása

Hozzon létre egy táblát a streambetöltési adatok fogadásához, és definiálja a kapcsolódó szabályzatot a Azure Synapse Studio vagy a Azure Portal használatával.

Studio

1. A Synapse Studio bal oldali ablaktábláján válassza a Fejlesztés lehetőséget.

2. A KQL-szkriptek területen válassza az + (Új erőforrás hozzáadása) >KQL-szkript lehetőséget. A jobb oldali panelen elnevezheti a szkriptet.

3. A Csatlakozás menüben válassza a contosodataexplorer elemet.

4. Az Adatbázis használata menüben válassza a TestDatabase lehetőséget.

5. Illessze be a következő parancsot, és válassza a Futtatás lehetőséget a tábla létrehozásához.

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

6. Másolja az alábbi parancsok egyikét a Lekérdezés panelre , és válassza a Futtatás lehetőséget. Ez határozza meg a streamelési betöltési szabályzatot a létrehozott táblában vagy a táblát tartalmazó adatbázisban.

   Tipp

   Az adatbázis szintjén definiált szabályzat az adatbázis összes meglévő és jövőbeli táblájára vonatkozik.

   • A létrehozott táblára vonatkozó szabályzat meghatározásához használja a következőt:

     .alter table TestTable policy streamingingestion enable
     

   • A létrehozott táblát tartalmazó adatbázis szabályzatának meghatározásához használja a következőt:

     .alter database StreamingTestDb policy streamingingestion enable
     

Portál

1. Az Azure Synapse Analytics portálon lépjen a Data Explorer készletre.

2. Válassza a Lekérdezés lehetőséget.

3. Ha streambetöltéssel szeretné létrehozni az adatokat fogadó táblát, másolja a következő parancsot a Lekérdezés panelre , és válassza a Futtatás lehetőséget.

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

4. Másolja az alábbi parancsok egyikét a Lekérdezés panelre , és válassza a Futtatás lehetőséget. Ez határozza meg a streamelési betöltési szabályzatot a létrehozott táblában vagy a táblát tartalmazó adatbázisban.

   Tipp

   Az adatbázis szintjén definiált szabályzat az adatbázis összes meglévő és jövőbeli táblájára vonatkozik.

   • A létrehozott táblára vonatkozó szabályzat meghatározásához használja a következőt:

     .alter table TestTable policy streamingingestion enable
     

   • A létrehozott táblát tartalmazó adatbázis szabályzatának meghatározásához használja a következőt:

     .alter database StreamingTestDb policy streamingingestion enable
     

Streamelési adatbetöltési alkalmazás létrehozása a Data Explorer-készletbe való adatbetöltéshez

Hozza létre az alkalmazást az adatok Data Explorer készletbe való betöltéséhez az előnyben részesített nyelv használatával. A poolPath változóhoz használja azt a lekérdezési végpontot, amelyről az Előfeltételek területen megjegyzést tett.

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

Ugrás

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

Streambetöltés letiltása a Data Explorer-készleten

Figyelmeztetés

A streambetöltés letiltása eltarthat néhány óráig.

Mielőtt letiltja a streambetöltést a Data Explorer-készleten, elvetheti a streamelési betöltési szabályzatot az összes releváns táblából és adatbázisból. A streamelési betöltési szabályzat eltávolítása adatáthelyezést vált ki a Data Explorer készleten belül. A streamelési betöltési adatok átkerülnek a kezdeti tárolóból az oszloptároló állandó tárolójába (kiterjedések vagy szegmensek). Ez a folyamat a kezdeti tárolóban lévő adatok mennyiségétől függően néhány másodperctől néhány óráig is eltarthat.

A streamelési betöltési szabályzat elvetése

A streamelési betöltési szabályzatot elvetheti a Azure Synapse Studióval vagy a Azure Portal.

Studio

1. A Synapse Studio bal oldali ablaktábláján válassza a Fejlesztés lehetőséget.

2. A KQL-szkriptek területen válassza az + (Új erőforrás hozzáadása) >KQL-szkript lehetőséget. A jobb oldali panelen elnevezheti a szkriptet.

3. A Csatlakozás menüben válassza a contosodataexplorer elemet.

4. Az Adatbázis használata menüben válassza a TestDatabase lehetőséget.

5. Illessze be a következő parancsot, és válassza a Futtatás lehetőséget a tábla létrehozásához.

   .delete table TestTable policy streamingingestion
   

6. A Azure Portal lépjen a Data Explorer készletre.

7. A Beállítások területen válassza a Konfigurációk lehetőséget.

8. A Konfigurációk panelen válassza a Be lehetőséget a streambetöltés engedélyezéséhez.

9. Válassza a Mentés lehetőséget.

Portál

1. A Azure Portal lépjen a Data Explorer készletre, és válassza a Lekérdezés lehetőséget.

2. Ha el szeretné dobni a streamelési betöltési szabályzatot a táblából, másolja a következő parancsot a Lekérdezés panelre , és válassza a Futtatás lehetőséget.

   .delete table TestTable policy streamingingestion
   

3. A Beállítások területen válassza a Konfigurációk lehetőséget.

4. A Konfigurációk panelen válassza a Ki lehetőséget a streambetöltés letiltásához.

5. Válassza a Mentés lehetőséget.

Korlátozások

• Az adatbázis-kurzorok nem támogatottak az adatbázisok esetében, ha maga az adatbázis vagy annak bármelyik táblája rendelkezik a Streambetöltési szabályzat definiálva és engedélyezve van.
• Az adatleképezéseketelőre létre kell hozni a streambetöltéshez. Az egyes streamelési betöltési kérések nem fogadják el a beágyazott adatleképezéseket.
• A bővítménycímkék nem állíthatók be a streamelési betöltési adatokon.
• Szabályzat frissítése. A frissítési szabályzat csak a forrástáblában újonnan betöltött adatokra hivatkozhat, az adatbázisban lévő többi adatra vagy táblára nem.
• Ha a streambetöltést az adatbázis bármelyik tábláján használják, ez az adatbázis nem használható vezetőként követő adatbázisokhoz vagy adatszolgáltatóként Azure Synapse Analytics-Data Share számára.

Következő lépések

• Elemzés Data Explorer
• Data Explorer-készletek monitorozása