Konfigurowanie ingestowania strumieniowego w puli usługi Azure Synapse Data Explorer (wersja próbna)

Podczas tworzenia puli Eksploratora danych, wykonując kroki opisane w Tworzeniu puli eksploratora danych przy użyciu Synapse Studio, na karcie Dodatkowe ustawienia wybierz pozycję Ingestia strumieniowa>Włączone.

Podczas tworzenia puli Eksploratora danych, wykonując kroki opisane w temacie Tworzenie puli eksploratora danych przy użyciu witryny Azure Portal, na karcie Ustawienia dodatkowe wybierz pozycję pozyskiwanie przesyłania strumieniowego>Włączone.

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

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 politykę strumieniowego pozyskiwania danych w tabeli, którą utworzyłeś, lub w bazie danych zawierającej tę tabelę.

   Wskazówka

   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
     

1. W portalu usługi Azure Synapse Analytics przejdź do puli eksploratora danych.

2. Wybierz pozycję Query.

3. Aby utworzyć tabelę, która będzie odbierać dane za pomocą strumieniowego przesyłania danych, skopiuj następujące polecenie do okienka zapytań 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 politykę strumieniowego pozyskiwania danych w tabeli, którą utworzyłeś, lub w bazie danych zawierającej tę tabelę.

   Wskazówka

   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
     

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

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

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ę.

   .delete table TestTable policy streamingingestion
   

6. W portalu Azure Portal przejdź do puli eksploratora danych.

7. W obszarze Ustawienia wybierz pozycję Konfiguracje.

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

9. Wybierz Zapisz.

1. W portalu Azure przejdź do swojej puli Eksploratora Danych i wybierz Zapytanie.

2. Aby usunąć politykę przesyłania danych strumieniowych z tabeli, skopiuj następujące polecenie do panelu zapytań i kliknij Uruchom.

   .delete table TestTable policy streamingingestion
   

3. W obszarze Ustawienia wybierz pozycję Konfiguracje.

4. W okienku Konfiguracje wybierz pozycję Off, aby wyłączyć przesyłanie strumieniowe.

5. Wybierz Zapisz.