Streaming-opname configureren op uw Azure Data Explorer cluster

Terwijl u een cluster maakt met behulp van de stappen in Een cluster Azure Data Explorer databasemaken, selecteert u op het tabblad Configuraties de optie Streaming-opnameop.

Als u streaming-opname wilt inschakelen tijdens het maken van een Azure Data Explorer cluster, moet u de volgende code uitvoeren:

using System.Threading.Tasks;
using Microsoft.Azure.Management.Kusto; // Required package Microsoft.Azure.Management.Kusto
using Microsoft.Azure.Management.Kusto.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory; // Required package Microsoft.IdentityModel.Clients.ActiveDirectory
using Microsoft.Rest;
namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string clusterName = "<clusterName>";
            string resourceGroupName = "<resourceGroupName>";
            string subscriptionId = "<subscriptionId>";
            string location = "<location>";
            string skuName = "<skuName>";
            string tier = "<tier>";

            var authenticationContext = new AuthenticationContext($"https://login.windows.net/{appTenant}");
            var credential = new ClientCredential(appId, appKey);
            var result = await authenticationContext.AcquireTokenAsync(resource: "https://management.core.windows.net/", clientCredential: credential);

            var credentials = new TokenCredentials(result.AccessToken, result.AccessTokenType);
            var kustoManagementClient = new KustoManagementClient(credentials)
            {
                SubscriptionId = subscriptionId
            };

            var cluster = new Cluster(location, new AzureSku(skuName, tier), enableStreamingIngest:true);
            await kustoManagementClient.Clusters.CreateOrUpdateAsync(resourceGroupName, clusterName, cluster);
        }
    }
}

1. Ga in Azure Portal naar uw Azure Data Explorer cluster.

2. Selecteer Instellingenin Instellingen configuraties.

3. Selecteer in het deelvenster Configuraties de optie Aan omStreaming-opname in teschakelen.

4. Selecteer Opslaan.

   

U kunt streaming-opname inschakelen tijdens het maken van een nieuw Azure Data Explorer cluster.

using System.Threading.Tasks;
using Microsoft.Azure.Management.Kusto; // Required package Microsoft.Azure.Management.Kusto
using Microsoft.Azure.Management.Kusto.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory; // Required package Microsoft.IdentityModel.Clients.ActiveDirectory
using Microsoft.Rest;
namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string clusterName = "<clusterName>";
            string resourceGroupName = "<resourceGroupName>";
            string subscriptionId = "<subscriptionId>";

            var authenticationContext = new AuthenticationContext($"https://login.windows.net/{appTenant}");
            var credential = new ClientCredential(appId, appKey);
            var result = await authenticationContext.AcquireTokenAsync(resource: "https://management.core.windows.net/", clientCredential: credential);

            var credentials = new TokenCredentials(result.AccessToken, result.AccessTokenType);
            var kustoManagementClient = new KustoManagementClient(credentials)
            {
                SubscriptionId = subscriptionId
            };

            var clusterUpdateParameters = new ClusterUpdate(enableStreamingIngest: true);
            await kustoManagementClient.Clusters.UpdateAsync(resourceGroupName, clusterName, clusterUpdateParameters);
        }
    }
}

1. Navigeer Azure Portal het cluster.

2. Selecteer Query.

   

3. Als u de tabel wilt maken die de gegevens ontvangt via streaming-opname, 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 streaming-opnamebeleid voor de tabel die u hebt gemaakt of op de database die de tabel bevat.

   Tip

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

   • Gebruik het volgende om het beleid te definiëren voor de tabel die u hebt gemaakt:

     .alter table TestTable policy streamingingestion enable
     

   • Gebruik het volgende om het beleid te definiëren voor de database met de tabel die u hebt gemaakt:

     .alter database StreamingTestDb policy streamingingestion enable
     

   

using System.Threading.Tasks;
using Kusto.Data; // Requires Package Microsoft.Azure.Kusto.Data
using Kusto.Data.Common;
using Kusto.Data.Net.Client;
namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string clusterPath = "https://<clusterName>.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(clusterPath)
                    .WithAadApplicationKeyAuthentication(
                        applicationClientId: appId,
                        applicationKey: appKey,
                        authority: appTenant
                    );

            var tableSchema = new TableSchema(
                tableName,
                new ColumnSchema[]
                {
                    new ColumnSchema("TimeStamp", "System.DateTime"),
                    new ColumnSchema("Name", "System.String"),
                    new ColumnSchema("Metric", "System.int"),
                    new ColumnSchema("Source", "System.String"),
                });
            var tableCreateCommand = CslCommandGenerator.GenerateTableCreateCommand(tableSchema);
            var tablePolicyAlterCommand = CslCommandGenerator.GenerateTableAlterStreamingIngestionPolicyCommand(tableName, isEnabled: true);

            using (var client = KustoClientFactory.CreateCslAdminProvider(csb))
            {
                client.ExecuteControlCommand(tableCreateCommand);
                client.ExecuteControlCommand(tablePolicyAlterCommand);
            }
        }
    }
}

using Kusto.Data;
using Kusto.Ingest;
using System.IO;
using Kusto.Data.Common;

namespace StreamingIngestion
{
    class Program
    {
        static void Main(string[] args)
        {
            string clusterPath = "https://<clusterName>.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(clusterPath)
                    .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.kusto.data import KustoConnectionStringBuilder

from azure.kusto.ingest import (
    IngestionProperties,
    DataFormat,
    KustoStreamingIngestClient
)

clusterPath = "https://<clusterName>.kusto.windows.net"
appId = "<appId>"
appKey = "<appKey>"
appTenant = "<appTenant>"
dbName = "<dbName>"
tableName = "<tableName>"

csb = KustoConnectionStringBuilder.with_aad_application_key_authentication(
    clusterPath,
    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 clusterPath = "https://<clusterName>.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(
        clusterPath,
        appId,
        appKey,
        appTenant
    ),
    ingestionProperties
);

// Automatically detects gz format
await client.ingestFromFile("MyFile.gz", ingestionProperties);

import (
    "context"
    "github.com/Azure/azure-kusto-go/kusto"
    "github.com/Azure/azure-kusto-go/kusto/ingest"
    "github.com/Azure/go-autorest/autorest/azure/auth"
)

func ingest() {
    clusterPath := "https://<clusterName>.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/go-autorest/autorest/azure/auth"
    authorizer := kusto.Authorization{
        Config: auth.NewClientCredentialsConfig(appId, appKey, appTenant),
    }

    // Create a client
    client, err := kusto.New(clusterPath, 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.kusto.data.auth.ConnectionStringBuilder;
import com.microsoft.azure.kusto.ingest.IngestClient;
import com.microsoft.azure.kusto.ingest.IngestClientFactory;
import com.microsoft.azure.kusto.ingest.IngestionProperties;
import com.microsoft.azure.kusto.ingest.result.OperationStatus;
import com.microsoft.azure.kusto.ingest.source.CompressionType;
import com.microsoft.azure.kusto.ingest.source.StreamSourceInfo;
import java.io.FileInputStream;
import java.io.InputStream;

public class FileIngestion {
    public static void main(String[] args) throws Exception {
        String clusterPath = "https://<clusterName>.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(
                clusterPath,
                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. Ga in Azure Portal naar uw Azure Data Explorer cluster en selecteer Query uitvoeren.

2. Als u het beleid voor streaming-opname 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 Instellingenin Instellingen configuraties.

4. Selecteer in het deelvenster Configuraties de optie Uit omStreaming-opname uit te schakelen.

5. Selecteer Opslaan.

   

Als u het beleid voor streaming-opname uit de tabel wilt neerzetten, moet u de volgende code uitvoeren:

using System.Threading.Tasks;
using Kusto.Data; // Requires Package Microsoft.Azure.Kusto.Data
using Kusto.Data.Common;
using Kusto.Data.Net.Client;
namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string clusterPath = "https://<clusterName>.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(clusterPath)
                    .WithAadApplicationKeyAuthentication(
                        applicationClientId: appId,
                        applicationKey: appKey,
                        authority: appTenant
                    );
            
            var tablePolicyDropCommand = CslCommandGenerator.GenerateTableStreamingIngestionPolicyDropCommand(dbName, tableName);
            using (var client = KustoClientFactory.CreateCslAdminProvider(csb))
            {
                client.ExecuteControlCommand(tablePolicyDropCommand);
            }
        }
    }
}

Als u streaming-opname in uw cluster wilt uitschakelen, moet u de volgende code uitvoeren:

using System.Threading.Tasks;
using Microsoft.Azure.Management.Kusto; // Required package Microsoft.Azure.Management.Kusto
using Microsoft.Azure.Management.Kusto.Models;
using Microsoft.IdentityModel.Clients.ActiveDirectory; // Required package Microsoft.IdentityModel.Clients.ActiveDirectory
using Microsoft.Rest;
namespace StreamingIngestion
{
    class Program
    {
        static async Task Main(string[] args)
        {
            string appId = "<appId>";
            string appKey = "<appKey>";
            string appTenant = "<appTenant>";
            string clusterName = "<clusterName>";
            string resourceGroupName = "<resourceGroupName>";
            string subscriptionId = "<subscriptionId>";
            
            var authenticationContext = new AuthenticationContext($"https://login.windows.net/{appTenant}");
            var credential = new ClientCredential(appId, appKey);
            var result = await authenticationContext.AcquireTokenAsync(resource: "https://management.core.windows.net/", clientCredential: credential);
            
            var credentials = new TokenCredentials(result.AccessToken, result.AccessTokenType);
            var kustoManagementClient = new KustoManagementClient(credentials)
            {
                SubscriptionId = subscriptionId
        };
        
            var clusterUpdateParameters = new ClusterUpdate(enableStreamingIngest: false);
            await kustoManagementClient.Clusters.UpdateAsync(resourceGroupName, clusterName, clusterUpdateParameters);
        }
    }
}