Streamelési adatbetöltés konfigurálása az Azure Data Explorer-fürtön
Cikk
A streambetöltés akkor hasznos az adatok betöltéséhez, ha kis késésre van szüksége a betöltés és a lekérdezés között. Fontolja meg a streambetöltés használatát a következő esetekben:
Egy másodpercnél rövidebb késésre van szükség.
Számos olyan tábla működési feldolgozásának optimalizálása érdekében, ahol az adatok adatfolyama az egyes táblákba viszonylag kicsi (másodpercenként néhány rekord), de a teljes adatbetöltési mennyiség magas (több ezer rekord másodpercenként).
Ha az adatok adatfolyama az egyes táblákba magas (óránként 4 GB-nál nagyobb), fontolja meg az üzenetsoros betöltés használatát.
A korábbi SDK-verziókon alapuló kódmintákért tekintse meg az archivált cikket.
Válassza ki a megfelelő streamelési betöltési típust
Két streamelési betöltési típus támogatott:
Betöltési típus
Description
Adatkapcsolat
Az Event Hubs, a IoT Hub és az Event Grid-adatkapcsolatok streambetöltést használhatnak, feltéve, hogy engedélyezve van a fürt szintjén. A streambetöltés használatára vonatkozó döntés a céltáblán konfigurált streamelési betöltési szabályzatnak megfelelően történik. Az adatkapcsolatok kezelésével kapcsolatos információkért lásd: Event Hub, IoT Hub és Event Grid.
Egyéni betöltés
Az egyéni betöltéshez olyan alkalmazást kell írnia, amely az Azure Data Explorer ügyfélkódtárak egyikét használja. Az ebben a témakörben található információk segítségével konfigurálhatja az egyéni betöltési adatokat. A C#-streamelési mintaalkalmazás is hasznos lehet.
Az alábbi táblázat segítségével kiválaszthatja a környezetének megfelelő betöltési típust:
Kritérium
Adatkapcsolat
Egyéni betöltés
Adatbetöltés kezdeményezése és a lekérdezéshez elérhető adatok közötti adatkésés
Hosszabb késleltetés
Rövidebb késleltetés
Fejlesztési többletterhelés
Gyors és egyszerű beállítás, nincs fejlesztési többletterhelés
Nagy fejlesztési terhelés az adatok betöltését, a hibák kezelését és az adatkonzisztenciát biztosító alkalmazások létrehozásához
Megjegyzés
A streambetöltés engedélyezésének és letiltásának folyamatát a Azure Portal vagy programozott módon c# nyelven kezelheti a fürtön. Ha az egyéni alkalmazáshoz C#-ot használ, a programozott megközelítéssel kényelmesebbnek találhatja.
Teljesítménnyel és működésével kapcsolatos szempontok
A streambetöltést befolyásoló fő közreműködők a következők:
Virtuális gép és fürt mérete: A streamelési betöltési teljesítmény és a kapacitás skálázható a virtuális gép és a fürt méretének növelésével. Az egyidejű betöltési kérelmek száma magonként legfeljebb hat lehet. Például a 16 magos termékváltozatok, például a D14 és az L16 esetében a maximálisan támogatott terhelés 96 egyidejű betöltési kérés. Két alapvető termékváltozat , például a D11 esetében a maximálisan támogatott terhelés 12 egyidejű betöltési kérelem.
Adatméretkorlát: A streamelési betöltési kérelem adatméretkorlátja 4 MB. Ide tartoznak a frissítési szabályzatokhoz a betöltés során létrehozott adatok.
Sémafrissítések: A sémafrissítések, például a táblák és a betöltési leképezések létrehozása és módosítása akár öt percet is igénybe vehet a streamelési szolgáltatás számára. További információ: Streambetöltés és sémamódosítások.
SSD-kapacitás: A streamelési adatbetöltés engedélyezése egy fürtön még akkor is, ha az adatok nem streameléssel vannak betöltve, a fürtgépek helyi SSD-lemezének egy részét használja a betöltési adatok streamelésére, és csökkenti a gyorsgyorsítótár számára elérhető tárhelyet.
Ha engedélyezni szeretné a streambetöltést egy új Azure Data Explorer-fürt létrehozásakor, futtassa a következő kódot:
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);
}
}
Streambetöltés engedélyezése meglévő fürtön
Ha már rendelkezik fürtel, engedélyezheti a streambetöltést a Azure Portal vagy programozott módon C# nyelven.
A Azure Portal lépjen az Azure Data Explorer-fürtre.
A Beállítások területen válassza a Konfigurációk lehetőséget.
A Konfigurációk panelen válassza a Be lehetőséget a streambetöltés engedélyezéséhez.
Kattintson a Mentés gombra.
Engedélyezheti a streambetöltést egy meglévő Azure Data Explorer-fürt frissítésekor.
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);
}
}
Céltábla létrehozása és a szabályzat meghatározása
Hozzon létre egy táblát a streambetöltési adatok fogadásához, és definiálja a kapcsolódó szabályzatot a Azure Portal vagy programozott módon C# nyelven.
Ha streambetöltéssel szeretné létrehozni az adatokat fogadó táblát, másolja a következő parancsot a Lekérdezés panelre , és válassza a Futtatás lehetőséget.
Másolja az alábbi parancsok egyikét a Lekérdezés panelre , és válassza a Futtatás lehetőséget. Ez határozza meg a streamelési betöltési szabályzatot a létrehozott táblában vagy a táblát tartalmazó adatbázisban.
Tipp
Az adatbázis szintjén definiált szabályzat az adatbázis összes meglévő és jövőbeli táblájára vonatkozik. Ha az adatbázis szintjén engedélyezi a szabályzatot, nem szükséges táblánként engedélyeznie.
A létrehozott táblára vonatkozó szabályzat meghatározásához használja a következőt:
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);
}
}
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")
}
}
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;
}
}
Streambetöltés letiltása a fürtön
Figyelmeztetés
A streambetöltés letiltása eltarthat néhány óráig.
Mielőtt letiltja a streambetöltést az Azure Data Explorer-fürtön, elvetheti a streamelési betöltési szabályzatot az összes releváns táblából és adatbázisból. A streamelési betöltési szabályzat eltávolítása adatáthelyezést vált ki az Azure Data Explorer-fürtön belül. A streamelési betöltési adatok átkerülnek a kezdeti tárolóból az oszloptároló állandó tárolójába (kiterjedések vagy szegmensek). Ez a folyamat a kezdeti tárolóban lévő adatok mennyiségétől függően néhány másodperctől néhány óráig is eltarthat.
A streamelési betöltési szabályzat elvetése
A streamelési betöltési szabályzatot elvetheti a Azure Portal vagy programozott módon C# nyelven.
A Azure Portal lépjen az Azure Data Explorer-fürtre, és válassza a Lekérdezés lehetőséget.
Ha el szeretné dobni a streamelési betöltési szabályzatot a táblából, másolja a következő parancsot a Lekérdezés panelre , és válassza a Futtatás lehetőséget.
.delete table TestTable policy streamingingestion
A Beállítások területen válassza a Konfigurációk lehetőséget.
A Konfigurációk panelen válassza a Ki lehetőséget a streambetöltés letiltásához.
Kattintson a Mentés gombra.
A streamelési betöltési szabályzat táblából való elvetéséhez futtassa a következő kódot:
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);
}
}
Ha le szeretné tiltani a streambetöltést a fürtön, futtassa a következő kódot:
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);
}
}
A bővítménycímkék nem állíthatók be a streamelési betöltési adatokon.
Szabályzat frissítése. A frissítési szabályzat csak a forrástábla újonnan betöltött adataira hivatkozhat, az adatbázisban lévő többi adatra vagy táblára nem.
Ha a streambetöltés engedélyezve van egy olyan fürtön, amely vezető szerepet tölt be a követő adatbázisokban, a streambetöltést a következő fürtökön is engedélyezni kell a streamelési betöltési adatok követéséhez. Ugyanez vonatkozik arra is, hogy a fürt adatai Data Share keresztül legyenek-e megosztva.