Configuração da ingestão de streaming no pool do Azure Synapse Data Explorer (Versão Prévia)

Ao criar um pool do Data Explorer usando as etapas em Criar um pool do Data Explorer usando o Synapse Studio, na guia Configurações adicionais, selecione Ingestão de> streaminghabilitada.

Ao criar um pool do Data Explorer usando as etapas em Criar um pool do Data Explorer usando o portal do Azure, na guia Configurações adicionais, selecione Ingestão de> streaminghabilitada.

1. No Synapse Studio, no painel esquerdo, selecione Desenvolver.

2. Em Scripts KQL, Selecione + (Adicionar novo recurso) >Script KQL. No painel do lado direito, você pode nomear seu script.

3. No menu Conectar a, selecione contosodataexplorer.

4. No menu Usar banco de dados, selecione TestDatabase.

5. Cole no comando a seguir e selecione Executar para criar a tabela.

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

6. Copie um dos seguintes comandos para o painel Consulta e selecione Executar. Isso define a política de ingestão de streaming na tabela que você criou ou no banco de dados que contém a tabela.

   Sugestão

   Uma política definida no nível do banco de dados se aplica a todas as tabelas existentes e futuras no banco de dados.

   • Para definir a política na tabela que você criou, use:

     .alter table TestTable policy streamingingestion enable
     

   • Para definir a política no banco de dados que contém a tabela que você criou, use:

     .alter database StreamingTestDb policy streamingingestion enable
     

1. No portal do Azure Synapse Analytics, navegue até o pool do Data Explorer.

2. Selecione Consulta.

3. Para criar a tabela que receberá os dados por ingestão de streaming, copie o seguinte comando no painel Consulta e selecione Executar.

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

4. Copie um dos seguintes comandos para o painel Consulta e selecione Executar. Isso define a política de ingestão de streaming na tabela que você criou ou no banco de dados que contém a tabela.

   Sugestão

   Uma política definida no nível do banco de dados se aplica a todas as tabelas existentes e futuras no banco de dados.

   • Para definir a política na tabela que você criou, use:

     .alter table TestTable policy streamingingestion enable
     

   • Para definir a política no banco de dados que contém a tabela que você criou, use:

     .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. No Synapse Studio, no painel esquerdo, selecione Desenvolver.

2. Em Scripts KQL, Selecione + (Adicionar novo recurso) >Script KQL. No painel do lado direito, você pode nomear seu script.

3. No menu Conectar a, selecione contosodataexplorer.

4. No menu Usar banco de dados, selecione TestDatabase.

5. Cole no comando a seguir e selecione Executar para criar a tabela.

   .delete table TestTable policy streamingingestion
   

6. No portal do Azure, vá para o pool do Data Explorer.

7. Em Configurações, selecione Configurações.

8. No painel Configurações , selecione Ativado para ativar a ingestão de streaming.

9. Selecione Guardar.

1. No portal do Azure, vá para o pool do Data Explorer e selecione Consulta.

2. Para soltar a política de ingestão de streaming da tabela, copie o seguinte comando no painel Consulta e selecione Executar.

   .delete table TestTable policy streamingingestion
   

3. Em Configurações, selecione Configurações.

4. No painel Configurações , selecione Desativado para desativar a ingestão de streaming.

5. Selecione Guardar.