Streaming-ingestie configureren in uw Azure Synapse Data Explorer-pool (Preview)

Wanneer u een Data Explorer-pool maakt met behulp van de stappen in Een Data Explorer-pool maken met behulp van Synapse Studio, selecteert u streamingopname> op het tabblad Aanvullende instellingen.

Wanneer u een Data Explorer-pool maakt met behulp van de stappen in Een Data Explorer-pool maken met behulp van De Azure-portal, selecteert u op het tabblad Aanvullende instellingen de optie Streamingopname>ingeschakeld.

1. Selecteer Ontwikkelen in Synapse Studio in het linkerdeelvenster.

2. Onder KQL-scripts selecteer + (nieuwe resource toevoegen) >KQL-script. In het rechterdeelvenster kunt u uw script een naam geven.

3. Selecteer contosodataexplorer in het menu Verbinding maken.

4. Selecteer in het menu Database gebruikenTestDatabase.

5. Plak de volgende opdracht en selecteer Uitvoeren om de tabel te maken.

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

6. Kopieer een van de volgende opdrachten naar het deelvenster Query en selecteer Uitvoeren. Hiermee definieert u het streamingopnamebeleid voor de tabel die u hebt gemaakt of in de database die de tabel bevat.

   Aanbeveling

   Een beleid dat op databaseniveau is gedefinieerd, is van toepassing op alle bestaande en toekomstige tabellen in de database.

   • Als u het beleid wilt definiëren voor de tabel die u hebt gemaakt, gebruikt u:

     .alter table TestTable policy streamingingestion enable
     

   • Als u het beleid wilt definiëren voor de database met de tabel die u hebt gemaakt, gebruikt u:

     .alter database StreamingTestDb policy streamingingestion enable
     

1. Navigeer in de Azure Synapse Analytics-portal naar uw Data Explorer-pool.

2. Selecteer Query.

3. Als u de tabel wilt maken die de gegevens ontvangt via streamingopname, kopieert u de volgende opdracht naar het deelvenster Query en selecteert u Uitvoeren.

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

4. Kopieer een van de volgende opdrachten naar het deelvenster Query en selecteer Uitvoeren. Hiermee definieert u het streamingopnamebeleid voor de tabel die u hebt gemaakt of in de database die de tabel bevat.

   Aanbeveling

   Een beleid dat op databaseniveau is gedefinieerd, is van toepassing op alle bestaande en toekomstige tabellen in de database.

   • Als u het beleid wilt definiëren voor de tabel die u hebt gemaakt, gebruikt u:

     .alter table TestTable policy streamingingestion enable
     

   • Als u het beleid wilt definiëren voor de database met de tabel die u hebt gemaakt, gebruikt u:

     .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. Selecteer Ontwikkelen in Synapse Studio in het linkerdeelvenster.

2. Onder KQL-scripts selecteer + (nieuwe resource toevoegen) >KQL-script. In het rechterdeelvenster kunt u uw script een naam geven.

3. Selecteer contosodataexplorer in het menu Verbinding maken.

4. Selecteer in het menu Database gebruikenTestDatabase.

5. Plak de volgende opdracht en selecteer Uitvoeren om de tabel te maken.

   .delete table TestTable policy streamingingestion
   

6. Ga in Azure Portal naar uw Data Explorer-pool.

7. Selecteer Configuraties in Instellingen.

8. Selecteer in het deelvenster Configuraties de optie Aan om streamingopname in te schakelen.

9. Selecteer Opslaan.

1. Ga in Azure Portal naar uw Data Explorer-pool en selecteer Query.

2. Als u het streamingopnamebeleid uit de tabel wilt verwijderen, kopieert u de volgende opdracht naar het deelvenster Query en selecteert u Uitvoeren.

   .delete table TestTable policy streamingingestion
   

3. Selecteer Configuraties in Instellingen.

4. Selecteer in het deelvenster Configuraties de optie Uit om streamingopname uit te schakelen.

5. Selecteer Opslaan.