Azure Veri Gezgini kümenizde akış veri alımını yapılandırma

Akış alımı, veri alımı ile sorgu arasında düşük gecikme süresine ihtiyaç duyduğunuzda verileri yüklemek için kullanışlıdır. Aşağıdaki senaryolarda akış alımını kullanmayı göz önünde bulundurun:

• Bir saniyeden kısa gecikme süresi gereklidir.
• Her tabloya veri akışının görece küçük olduğu (saniye başına birkaç kayıt) ancak genel veri alımı hacminin yüksek olduğu (saniyede binlerce kayıt) birçok tablonun operasyonel işlemesini iyileştirmek için.

Her tabloya veri akışı yüksekse (saatte 4 GB'ın üzerinde), toplu alma kullanmayı göz önünde bulundurun.

Farklı alım yöntemleri hakkında daha fazla bilgi edinmek için bkz. Veri alımına genel bakış.

Uygun akış alımı türünü seçin

İki akış alımı türü desteklenir:

Alım türü	Açıklama
Veri bağlantısı	Event Hub, IoT Hub ve Event Grid veri bağlantıları, küme düzeyinde etkinleştirilmiş olması koşuluyla akış alımını kullanabilir. Akış alımını kullanma kararı, hedef tabloda yapılandırılan akış alma ilkesine göre yapılır. Veri bağlantılarını yönetme hakkında bilgi için bkz. Event Hub, IoT Hub ve Event Grid.
Özel alma	Özel alım, Azure Veri Gezgini istemci kitaplıklarından birini kullanan bir uygulama yazmanızı gerektirir. Özel alımı yapılandırmak için bu konudaki bilgileri kullanın. C# akış alımı örnek uygulamasını da yararlı bulabilirsiniz.

Ortamınıza uygun alım türünü seçmenize yardımcı olması için aşağıdaki tabloyu kullanın:

Ölçütü	Veri bağlantısı	Özel Alma
Veri alımı başlatma ile sorgu için kullanılabilir veriler arasındaki veri gecikmesi	Daha uzun gecikme	Daha kısa gecikme
Geliştirme yükü	Hızlı ve kolay kurulum, geliştirme yükü yok	Bir uygulama oluşturmak için yüksek geliştirme yükü verileri alır, hataları işler ve veri tutarlılığını güvence altına alır

Not

Azure portal kullanarak veya C# dilinde program aracılığıyla kümenizde akış alımını etkinleştirme ve devre dışı bırakma işlemini yönetebilirsiniz. Özel uygulamanız için C# kullanıyorsanız, programlama yaklaşımını kullanarak daha kullanışlı bulabilirsiniz.

Önkoşullar

• Azure aboneliği. Ücretsiz bir Azure hesabı oluşturun.

Performans ve operasyonel konular

Akış alımını etkileyebilecek başlıca katkıda bulunanlar şunlardır:

• VM ve küme boyutu: Akış alımı performansı ve kapasite ölçekleri, artan VM ve küme boyutlarıyla sağlanır. Eş zamanlı alım isteklerinin sayısı çekirdek başına altı ile sınırlıdır. Örneğin, D14 ve L16 gibi 16 çekirdekli SKU'lar için desteklenen en yüksek yük 96 eşzamanlı alma isteğidir. D11 gibi iki çekirdek SKU için desteklenen en yüksek yük 12 eşzamanlı alma isteğidir.
• Veri boyutu sınırı: Akış alma isteğinin veri boyutu sınırı 4 MB'tır. Bu, veri alımı sırasında güncelleştirme ilkeleri için oluşturulan tüm verileri içerir.
• Şema güncelleştirmeleri: Tabloların oluşturulması ve değiştirilmesi ve alım eşlemeleri gibi şema güncelleştirmeleri, akış alma hizmeti için beş dakikaya kadar sürebilir. Daha fazla bilgi için bkz . Akış alımı ve şema değişiklikleri.
• SSD kapasitesi: Akış yoluyla veri alınmasa bile bir kümede akış alımını etkinleştirmek, veri akışı için küme makinelerinin yerel SSD diskinin bir bölümünü kullanır ve sık erişimli önbellek için kullanılabilir depolama alanını azaltır.

Kümenizde akış alımını etkinleştirme

Akış alımını kullanabilmeniz için önce kümenizde özelliği etkinleştirmeniz ve bir akış alımı ilkesi tanımlamanız gerekir. Kümeyi oluştururken özelliği etkinleştirebilir veya mevcut bir kümeye ekleyebilirsiniz.

Uyarı

Akış alımını etkinleştirmeden önce sınırlamaları gözden geçirin.

Yeni küme oluştururken akış alımını etkinleştirme

Azure portal kullanarak veya C# dilinde program aracılığıyla yeni bir küme oluştururken akış alımını etkinleştirebilirsiniz.

Portal

Azure Veri Gezgini kümesi ve veritabanı oluşturma adımlarını kullanarak küme oluştururken, Yapılandırmalar sekmesinde Akış alımı>Açık'ı seçin.

C#

Yeni bir Azure Veri Gezgini kümesi oluştururken akış alımını etkinleştirmek için aşağıdaki kodu çalıştırın:

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

Mevcut kümede akış alımını etkinleştirme

Mevcut bir kümeniz varsa, Azure portal kullanarak veya C# dilinde program aracılığıyla akış alımını etkinleştirebilirsiniz.

Portal

1. Azure portal Azure Veri Gezgini kümenize gidin.

2. Ayarlar'daYapılandırmalar'ı seçin.

3. Yapılandırmalar bölmesinde Açık'ı seçerek Akış alımını etkinleştirin.

4. Kaydet’i seçin.

   

C#

Yeni bir Azure Veri Gezgini kümesi oluştururken akış alımını etkinleştirebilirsiniz.

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

Hedef tablo oluşturma ve ilkeyi tanımlama

Akış alma verilerini almak için bir tablo oluşturun ve Azure portal kullanarak veya C# dilinde program aracılığıyla ilgili ilkesini tanımlayın.

Portal

1. Azure portal kümenize gidin.

2. Sorgu'yu seçin.

   

3. Akış alımı yoluyla verileri alacak tabloyu oluşturmak için aşağıdaki komutu Sorgu bölmesine kopyalayın ve Çalıştır'ı seçin.

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

   

4. Aşağıdaki komutlardan birini Sorgu bölmesine kopyalayın ve Çalıştır'ı seçin. Bu, oluşturduğunuz tabloda veya tabloyu içeren veritabanında akış alma ilkesini tanımlar.

   İpucu

   Veritabanı düzeyinde tanımlanan bir ilke, veritabanındaki tüm mevcut ve gelecekteki tablolar için geçerlidir. İlkeyi veritabanı düzeyinde etkinleştirdiğinizde tablo başına etkinleştirmeniz gerekmez.

   • Oluşturduğunuz tabloda ilkeyi tanımlamak için şunu kullanın:

     .alter table TestTable policy streamingingestion enable
     

   • Oluşturduğunuz tabloyu içeren veritabanında ilkeyi tanımlamak için şunu kullanın:

     .alter database StreamingTestDb policy streamingingestion enable
     

   

C#

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)
    {
        var clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        // Create Kusto connection string with App Authentication
        var connectionStringBuilder = new KustoConnectionStringBuilder(clusterPath)
            .WithAadApplicationKeyAuthentication(
                applicationClientId: appId,
                applicationKey: appKey,
                authority: appTenant
            );
        using var client = KustoClientFactory.CreateCslAdminProvider(connectionStringBuilder);
        
        var tableName = "<tableName>";
        var tableSchema = new TableSchema(
            tableName,
            new ColumnSchema[]
            {
                new("TimeStamp", "System.DateTime"),
                new("Name", "System.String"),
                new("Metric", "System.int"),
                new("Source", "System.String"),
            });
        var tableCreateCommand = CslCommandGenerator.GenerateTableCreateCommand(tableSchema);
        var tablePolicyAlterCommand = CslCommandGenerator.GenerateTableAlterStreamingIngestionPolicyCommand(tableName, isEnabled: true);
        
        await client.ExecuteControlCommandAsync(tableCreateCommand);
        await client.ExecuteControlCommandAsync(tablePolicyAlterCommand);
    }
}

Kümenize veri almak için akış alma uygulaması oluşturma

Tercih ettiğiniz dili kullanarak kümenize veri almak için uygulamanızı oluşturun.

C#

using System.IO;
using System.Threading.Tasks;
using Kusto.Data; // Requires Package Microsoft.Azure.Kusto.Data
using Kusto.Data.Common;
using Kusto.Ingest; // Requires Package Microsoft.Azure.Kusto.Ingest

namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        // Create Kusto connection string with App Authentication
        var connectionStringBuilder = new KustoConnectionStringBuilder(clusterPath)
            .WithAadApplicationKeyAuthentication(
                applicationClientId: appId,
                applicationKey: appKey,
                authority: appTenant
            );
        // Create a disposable client that will execute the ingestion
        using var client = KustoIngestFactory.CreateStreamingIngestClient(connectionStringBuilder);
        // Ingest from a compressed file
        var fileStream = File.Open("MyFile.gz", FileMode.Open);
        // Initialize client properties
        var ingestionProperties = new KustoIngestionProperties(databaseName: "<databaseName>", tableName: "<tableName>");
        // Create source options
        var sourceOptions = new StreamSourceOptions { CompressionType = DataSourceCompressionType.GZip, };
        // Ingest from stream
        await client.IngestFromStreamAsync(fileStream, ingestionProperties, sourceOptions);
    }
}

Python

from azure.kusto.data import KustoConnectionStringBuilder, DataFormat
from azure.kusto.ingest import IngestionProperties, KustoStreamingIngestClient

clusterPath = "https://<clusterName>.<region>.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) 

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 clusterPath = "https://<clusterName>.<region>.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);

Git

import (
    "context"
    "github.com/Azure/azure-kusto-go/kusto"
    "github.com/Azure/azure-kusto-go//azure/data-explorer/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")
    }
}

Java

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

Kümenizde akış alımını devre dışı bırakma

Uyarı

Akış alımının devre dışı bırakılması birkaç saat sürebilir.

Azure Veri Gezgini kümenizde akış alımını devre dışı bırakmadan önce, tüm ilgili tablolardan ve veritabanlarından akış alma ilkesini bırakın. Akış alma ilkesinin kaldırılması, Azure Veri Gezgini kümenizin içinde veri yeniden düzenlemesini tetikler. Akış alımı verileri ilk depolama alanından sütun deposundaki kalıcı depolama alanına taşınır (kapsamlar veya parçalar). Bu işlem, ilk depolamadaki veri miktarına bağlı olarak birkaç saniye ile birkaç saat arasında sürebilir.

Akış alma ilkesini bırakma

Azure portal kullanarak veya C# dilinde program aracılığıyla akış alma ilkesini bırakabilirsiniz.

Portal

1. Azure portal Azure Veri Gezgini kümenize gidin ve Sorgu'yu seçin.

2. Akış alma ilkesini tablodan bırakmak için aşağıdaki komutu Sorgu bölmesine kopyalayın ve Çalıştır'ı seçin.

   .delete table TestTable policy streamingingestion
   

   

3. Ayarlar'daYapılandırmalar'ı seçin.

4. Yapılandırmalar bölmesinde Kapalı'yı seçerek Akış alımını devre dışı bırakın.

5. Kaydet’i seçin.

   

C#

Akış alma ilkesini tablodan bırakmak için aşağıdaki kodu çalıştırın:

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)
    {
        var clusterPath = "https://<clusterName>.<region>.kusto.windows.net";
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        // Create Kusto connection string with App Authentication
        var connectionStringBuilder = new KustoConnectionStringBuilder(clusterPath)
            .WithAadApplicationKeyAuthentication(
                applicationClientId: appId,
                applicationKey: appKey,
                authority: appTenant
            );
        using var client = KustoClientFactory.CreateCslAdminProvider(connectionStringBuilder);
        var tablePolicyDropCommand = CslCommandGenerator.GenerateTableStreamingIngestionPolicyDropCommand("<dbName>", "<tableName>");
        await client.ExecuteControlCommandAsync(tablePolicyDropCommand);
    }
}

Kümenizde akış alımını devre dışı bırakmak için aşağıdaki kodu çalıştırın:

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

Sınırlamalar

• Veri eşlemeleri , akış alımında kullanılmak üzere önceden oluşturulmalıdır . Tek tek akış alımı istekleri satır içi veri eşlemelerini barındırmaz.
• Kapsam etiketleri akış alım verilerinde ayarlanamaz.
• İlkeyi güncelleştirin. Güncelleştirme ilkesi, veritabanındaki diğer verilere veya tablolara değil, yalnızca kaynak tablodaki yeni alınan verilere başvurabilir.
• Takip eden veritabanları için öncü olarak kullanılan bir kümede akış alımı etkinleştirildiyse, akış alma verilerinin takip edilebilmesi için aşağıdaki kümelerde akış alımının etkinleştirilmesi gerekir. Aynı durum, küme verilerinin Veri Paylaşımı aracılığıyla paylaşılıp paylaşılmadığı için de geçerlidir.

Sonraki adımlar

• Azure Veri Gezgini'de verileri sorgulama