Mengonfigurasi penyerapan streaming di kluster Azure Data Explorer Anda

Ingesti streaming berguna untuk memuat data ketika Anda membutuhkan latensi rendah antara ingesti dan query. Pertimbangkan untuk menggunakan konsumsi streaming dalam skenario berikut:

• Latensi kurang dari satu detik diperlukan.
• Untuk mengoptimalkan pemrosesan operasional banyak tabel di mana aliran data ke setiap tabel relatif kecil (beberapa catatan per detik), tetapi volume konsumsi data keseluruhan tinggi (ribuan catatan per detik).

Jika aliran data ke setiap tabel tinggi (lebih dari 4 GB per jam), pertimbangkan untuk menggunakan penyerapan yang diantrekan.

Untuk mempelajari lebih lanjut tentang metode konsumsi yang berbeda, lihat gambaran umum konsumsi data.

Untuk sampel kode berdasarkan versi SDK sebelumnya, lihat artikel yang diarsipkan.

Pilih jenis konsumsi streaming yang sesuai

Dua jenis konsumsi streaming didukung:

Jenis penyerapan	Deskripsi
Koneksi data	Koneksi data Azure Event Hubs, IoT Hub, dan Event Grid dapat menggunakan penyerapan streaming, asalkan diaktifkan pada tingkat kluster. Keputusan untuk menggunakan penyerapan streaming dilakukan sesuai dengan kebijakan penyerapan streaming yang dikonfigurasi pada tabel target. Untuk informasi tentang mengelola koneksi data, lihat Event Hub, IoT Hub, dan Event Grid.
Pemasukan Kustom	Penyerapan kustom mengharuskan Anda menulis aplikasi yang menggunakan salah satu pustaka klien Azure Data Explorer. Gunakan informasi dalam artikel ini untuk mengonfigurasi penyerapan kustom. Anda juga dapat menemukan aplikasi contoh penyerapan streaming C?view=azure-data-explorer&preserve-view=true# bermanfaat.

Gunakan tabel berikut untuk membantu Anda memilih jenis konsumsi yang sesuai untuk lingkungan Anda:

Kriteria	Koneksi data	Pemasukan Data Kustom
Keterlambatan data antara inisiasi konsumsi dan data yang tersedia untuk kueri	Penundaan yang lebih lama	Penundaan yang lebih pendek
Biaya overhead pengembangan	Pengaturan cepat dan mudah, tidak ada overhead pengembangan	Tingginya beban pengembangan untuk membuat aplikasi yang mengambil data, menangani kesalahan, dan memastikan konsistensi data.

Catatan

Anda dapat mengelola proses untuk mengaktifkan dan menonaktifkan penyerapan streaming pada kluster Anda menggunakan portal Azure atau secara terprogram di C#. Jika Anda menggunakan C# untuk aplikasi kustom, Anda mungkin merasa lebih nyaman menggunakan pendekatan terprogram.

Prasyarat

• Langganan Azure. Membuat akun Azure gratis.

Pertimbangan kinerja dan operasional

Kontributor utama yang dapat mempengaruhi konsumsi streaming adalah:

• Ukuran VM dan kluster: Performa penyerapan streaming dan kapasitas berkembang dengan peningkatan ukuran VM dan kluster. Jumlah permintaan konsumsi bersamaan dibatasi hingga enam per inti. Misalnya, untuk 16 SKU inti, seperti D14 dan L16, beban maksimal yang didukung adalah 96 permintaan penyerapan bersamaan. Untuk dua SKU inti, seperti D11, beban maksimal yang didukung adalah 12 permintaan penyerapan bersamaan.
• {4}Batas ukuran{2}data: Batas ukuran data untuk permintaan konsumsi streaming adalah 4 MB. Ini termasuk data apa pun yang dibuat untuk kebijakan pembaruan selama proses ingesti.
• Pembaruan skema:Pembaruan skema, seperti pembuatan dan modifikasi tabel dan pemetaan konsumsi, dapat memakan waktu hingga lima menit untuk layanan konsumsi streaming. Untuk informasi selengkapnya lihat Konsumsi streaming dan perubahan skema.
• Kapasitas SSD: Mengaktifkan penyerapan streaming pada kluster, bahkan ketika data tidak diserap melalui streaming, menggunakan bagian dari disk SSD lokal komputer kluster untuk data penyerapan streaming dan mengurangi penyimpanan yang tersedia untuk cache panas.
• Kursor database: Saat menggunakan penyerapan streaming, pembaruan kursor database mungkin tertinggal hingga 60 detik dari ketersediaan data. Penundaan ini timbul dari proses penyegelan latar belakang asinkron yang mentransisikan data dari buffer streaming ke ekstensi penyimpanan kolom permanen. Selama proses ini, kursor (yang digunakan untuk pemrosesan bertahap, ekspor berkelanjutan, atau tampilan materialisasi) diperbarui. Jika beban kerja Anda memerlukan konsistensi kursor secara segera untuk semantik tepat satu kali, pertimbangkan untuk menggunakan penerimaan berantrean sebagai gantinya atau perhitungkan potensi latensi ini dalam logika aplikasi Anda.

Mengaktifkan penyerapan streaming pada kluster Anda

Sebelum dapat menggunakan penyerapan streaming, Anda harus mengaktifkan kemampuan pada kluster Anda dan menentukan kebijakan penyerapan streaming. Anda dapat mengaktifkan kemampuan saat membuat kluster, atau menambahkannya ke kluster yang ada.

Peringatan

Tinjau batasan sebelum mengaktifkan konsumsi streaming.

Mengaktifkan penyerapan streaming saat membuat kluster baru

Anda dapat mengaktifkan penyerapan streaming saat membuat kluster baru menggunakan portal Azure atau secara terprogram di C#.

Portal

Saat membuat kluster menggunakan langkah-langkah dalam Membuat kluster dan database Azure Data Explorer, di tab Konfigurasi, pilih Ingesti streaming>Aktif.

C#

Untuk mengaktifkan penyerapan streaming saat membuat kluster Azure Data Explorer baru, jalankan kode berikut:

using System.Threading.Tasks;
using Azure;
using Azure.Core;
using Azure.Identity; // Required package Azure.Identity
using Azure.ResourceManager;
using Azure.ResourceManager.Kusto; // Required package Azure.ResourceManager.Kusto
using Azure.ResourceManager.Kusto.Models;
namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        var subscriptionId = "<subscriptionId>";
        var credentials = new ClientSecretCredential(appTenant, appId, appKey);
        var resourceManagementClient = new ArmClient(credentials, subscriptionId);
        var resourceGroupName = "<resourceGroupName>";
        var clusterName = "<clusterName>";
        
        var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
        var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
        var clusters = resourceGroup.GetKustoClusters();
        
        var location = new AzureLocation("<location>");
        var skuName = new KustoSkuName("<skuName>");
        var skuTier = new KustoSkuTier("<skuTier>");
        var clusterData = new KustoClusterData(location, new KustoSku(skuName, skuTier)) { IsStreamingIngestEnabled = true };
        await clusters.CreateOrUpdateAsync(WaitUntil.Completed, clusterName, clusterData);
    }
}

Mengaktifkan penyerapan streaming pada kluster yang ada

Jika Anda memiliki kluster yang ada, Anda dapat mengaktifkan penyerapan streaming menggunakan portal Azure atau secara terprogram di C#.

Portal

1. Di portal Azure, buka kluster Azure Data Explorer Anda.

2. Pada Pengaturan, pilih Konfigurasi.

3. Di panel Konfigurasi, pilih Aktif untuk mengaktifkan ingesti Streaming.

4. Pilih Simpan.

   

C#

Anda dapat mengaktifkan penyerapan streaming saat memperbarui kluster Azure Data Explorer yang ada.

using System.Threading.Tasks;
using Azure;
using Azure.Identity; // Required package Azure.Identity
using Azure.ResourceManager;
using Azure.ResourceManager.Kusto; // Required package Azure.ResourceManager.Kusto
using Azure.ResourceManager.Kusto.Models;
namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        var subscriptionId = "<subscriptionId>";
        var credentials = new ClientSecretCredential(appTenant, appId, appKey);
        var resourceManagementClient = new ArmClient(credentials, subscriptionId);
        var resourceGroupName = "<resourceGroupName>";
        var clusterName = "<clusterName>";
        var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
        var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
        var cluster = (await resourceGroup.GetKustoClusterAsync(clusterName)).Value;
        var clusterPatch = new KustoClusterPatch(cluster.Data.Location) { IsStreamingIngestEnabled = true };
        await cluster.UpdateAsync(WaitUntil.Completed, clusterPatch);
    }
}

Membuat tabel target dan menentukan kebijakan

Buat tabel untuk menerima data penyerapan streaming dan tentukan kebijakan terkait menggunakan portal Azure atau secara terprogram di C#.

Portal

1. Di portal Azure, buka kluster Anda.

2. Pilih Query.

   

3. Untuk membuat tabel yang akan menerima data melalui konsumsi streaming, salin perintah berikut ke panel Kueri dan pilih Jalankan.

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

   

4. Salin salah satu perintah berikut ke panel Kueri dan pilih Jalankan. Ini mendefinisikan kebijakan konsumsi streaming pada tabel yang Anda buat atau pada database yang berisi tabel.

   Petunjuk / Saran

   Kebijakan yang didefinisikan pada tingkat database berlaku untuk semua tabel yang ada dan masa depan dalam database. Saat Anda mengaktifkan kebijakan di tingkat database, anda tidak perlu mengaktifkannya per tabel.

   • Untuk menentukan kebijakan pada tabel yang Anda buat, gunakan:

     .alter table TestTable policy streamingingestion enable
     

   • Untuk menentukan kebijakan pada database yang berisi tabel yang Anda buat, gunakan:

     .alter database StreamingTestDb policy streamingingestion enable
     

   

C#

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

Membuat aplikasi penyerapan streaming untuk menyerap data ke kluster Anda

Buat aplikasi Anda untuk menyerap data ke kluster Anda menggunakan bahasa pilihan Anda.

Catatan

Untuk ingestasi antrean, lihat panduan untuk Membuat aplikasi untuk mendapatkan data menggunakan ingestasi antrean

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

Lanjut

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>.<region>.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")
    }
}

Jawa

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

Menonaktifkan penyerapan streaming pada kluster Anda

Peringatan

Menonaktifkan konsumsi streaming mungkin memakan waktu beberapa jam.

Sebelum menonaktifkan penyerapan streaming pada kluster Azure Data Explorer Anda, hilangkan kebijakan penyerapan streaming dari semua tabel dan database yang relevan. Penghapusan kebijakan penyerapan streaming memicu penataan ulang data di dalam kluster Azure Data Explorer Anda. Data pemasukan streaming dipindahkan dari penyimpanan awal ke penyimpanan permanen di penyimpanan kolom (ekstensi atau pecahan). Proses ini dapat memakan waktu antara beberapa detik hingga beberapa jam, tergantung pada jumlah data dalam penyimpanan awal.

Hapus kebijakan penerimaan streaming

Anda dapat menghilangkan kebijakan penyerapan streaming menggunakan portal Azure atau secara terprogram di C#.

Portal

1. Di portal Azure, buka kluster Azure Data Explorer Anda dan pilih Kueri.

2. Untuk menjatuhkan kebijakan konsumsi streaming dari tabel, salin perintah berikut ke panel Kueri dan pilih Jalankan.

   .delete table TestTable policy streamingingestion
   

   

3. Pada Pengaturan, pilih Konfigurasi.

4. Di panel Konfigurasi, pilih Mati untuk menonaktifkan Streaming ingestion.

5. Pilih Simpan.

   

C#

Untuk menghilangkan kebijakan penyerapan streaming dari tabel, jalankan kode berikut:

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

Untuk menonaktifkan penyerapan streaming pada kluster Anda, jalankan kode berikut:

using System.Threading.Tasks;
using Azure;
using Azure.Identity; // Required package Azure.Identity
using Azure.ResourceManager;
using Azure.ResourceManager.Kusto; // Required package Azure.ResourceManager.Kusto
using Azure.ResourceManager.Kusto.Models;
namespace StreamingIngestion;
class Program
{
    static async Task Main(string[] args)
    {
        var appId = "<appId>";
        var appKey = "<appKey>";
        var appTenant = "<appTenant>";
        var subscriptionId = "<subscriptionId>";
        var credentials = new ClientSecretCredential(appTenant, appId, appKey);
        var resourceManagementClient = new ArmClient(credentials, subscriptionId);
        var resourceGroupName = "<resourceGroupName>";
        var clusterName = "<clusterName>";
        var subscription = await resourceManagementClient.GetDefaultSubscriptionAsync();
        var resourceGroup = (await subscription.GetResourceGroupAsync(resourceGroupName)).Value;
        var cluster = (await resourceGroup.GetKustoClusterAsync(clusterName)).Value;
        var clusterPatch = new KustoClusterPatch(cluster.Data.Location) { IsStreamingIngestEnabled = false };
        await cluster.UpdateAsync(WaitUntil.Completed, clusterPatch);
    }
}

Batasan

• Pemetaan data harus dibuat sebelumnya untuk digunakan dalam konsumsi streaming. Permintaan konsumsi streaming individual tidak mengakomodasi pemetaan data inline.
• Tag ekstensi tidak dapat diatur pada data konsumsi streaming.
• Memperbarui kebijakan
  • Kebijakan pembaruan hanya dapat mereferensikan data yang baru diserap dalam tabel sumber dan bukan data atau tabel lain dalam database
  • Plugin Python tidak didukung
  • Ketika kebijakan pembaruan dengan kebijakan transaksi gagal, percobaan kembali ke penyerapan batch.
  • Untuk kebijakan pembaruan bertingkat yang menyertakan join operator, Anda harus menonaktifkan penyerapan streaming di semua tabel upstream. Misalnya, pertimbangkan kebijakan pembaruan berskala di mana Table1 memperbarui Table2, Table2 memperbarui Table3, dan Table3 memperbarui Table4. Jika kebijakan pembaruan Table4 menyertakan gabungan, Anda harus menonaktifkan penyerapan streaming pada Table1, Table2, dan Table3.
• Jika penyerapan streaming diaktifkan pada kluster yang digunakan sebagai pemimpin untuk database pengikut, penyerapan streaming harus diaktifkan pada kluster berikut juga untuk mengikuti data penyerapan streaming. Hal yang sama berlaku apakah data kluster dibagikan melalui Data Share.

Konten terkait

• Mengkueri data di Azure Data Explorer