Kusto .NET SDK'sını kullanarak veri alma

.NET için iki istemci kitaplığı vardır: alma kitaplığı ve veri kitaplığı. .NET SDK hakkında daha fazla bilgi için bkz . .NET SDK hakkında. Bu kitaplıklar verileri bir kümeye almanıza (yüklemenize ve kodunuzdan verileri sorgulamanıza olanak tanır. Bu makalede, önce bir test kümesinde bir tablo ve veri eşlemesi oluşturursunuz. Ardından bir alımı kümeye kuyruğa alır ve sonuçları doğrularsınız.

Önkoşullar

• Microsoft hesabı veya Microsoft Entra kullanıcı kimliği. Azure aboneliği gerekmez.
• Küme ve veritabanı. Küme ve veritabanı oluşturma.

Alma kitaplığını yükleme

Install-Package Microsoft.Azure.Kusto.Ingest

Kimlik doğrulaması ekleme ve bağlantı dizesi oluşturma

Kimlik Doğrulaması

Sdk, bir uygulamanın kimliğini doğrulamak için Microsoft Entra kiracı kimliğinizi kullanır. Kiracı kimliğinizi bulmak için aşağıdaki URL'yi kullanın ve YourDomain yerine kendi etki alanınızı yazın.

https://login.microsoftonline.com/<YourDomain>/.well-known/openid-configuration/

Örneğin, etki alanınız contoso.com olduğunda URL şöyle olur: https://login.microsoftonline.com/contoso.com/.well-known/openid-configuration/. Sonuçları görmek için bu URL'ye tıklayın; ilk satır aşağıdaki gibidir.

"authorization_endpoint":"https://login.microsoftonline.com/6babcaad-604b-40ac-a9d7-9fd97c0b779f/oauth2/authorize"

Bu örnekte kiracı kimliği 6babcaad-604b-40ac-a9d7-9fd97c0b779f değeridir.

Bu örnek, kümeye erişmek için etkileşimli bir Microsoft Entra kullanıcı kimlik doğrulaması kullanır. Microsoft Entra uygulama kimlik doğrulamayı sertifika veya uygulama gizli dizisiyle de kullanabilirsiniz. Bu kodu çalıştırmadan önce ve clusterUri için tenantId doğru değerleri ayarladığınızdan emin olun.

SDK, bağlantı dizesi bir parçası olarak kimlik doğrulama yöntemini ayarlamak için kullanışlı bir yol sağlar. bağlantı dizesi hakkında eksiksiz belgeler için bkz. bağlantı dizesi.

bağlantı dizesi oluşturma

Artık bağlantı dizesi oluşturabilirsiniz. Sonraki bir adımda hedef tabloyu ve eşlemeyi oluşturacaksınız.

var kustoUri = "https://<ClusterName>.<Region>.kusto.windows.net/";
var tenantId = "xxxxxxxx-xxxxx-xxxx-xxxx-xxxxxxxxx";
var kustoConnectionStringBuilder = new KustoConnectionStringBuilder(kustoUri).WithAadUserPromptAuthentication(tenantId);

Kaynak dosya bilgilerini ayarlama

Kaynak dosyanın yolunu ayarlayın. Bu örnekte, Azure Blob Depolama'da barındırılan bir örnek dosya kullanılır. StormEvents örnek veri kümesi, Ulusal Çevre Bilgileri Merkezlerinden hava durumuyla ilgili verileri içerir.

var blobPath = "https://kustosamples.blob.core.windows.net/samplefiles/StormEvents.csv";

Test kümenizde tablo oluşturma

Dosyadaki StormEvents.csv verilerin şemasıyla eşleşen adlı StormEvents bir tablo oluşturun.

var databaseName = "<DatabaseName>";
var tableName = "StormEvents";
using (var kustoClient = KustoClientFactory.CreateCslAdminProvider(kustoConnectionStringBuilder))
{
    var command = CslCommandGenerator.GenerateTableCreateCommand(
        tableName,
        new[]
        {
            Tuple.Create("StartTime", "System.DateTime"),
            Tuple.Create("EndTime", "System.DateTime"),
            Tuple.Create("EpisodeId", "System.Int32"),
            Tuple.Create("EventId", "System.Int32"),
            Tuple.Create("State", "System.String"),
            Tuple.Create("EventType", "System.String"),
            Tuple.Create("InjuriesDirect", "System.Int32"),
            Tuple.Create("InjuriesIndirect", "System.Int32"),
            Tuple.Create("DeathsDirect", "System.Int32"),
            Tuple.Create("DeathsIndirect", "System.Int32"),
            Tuple.Create("DamageProperty", "System.Int32"),
            Tuple.Create("DamageCrops", "System.Int32"),
            Tuple.Create("Source", "System.String"),
            Tuple.Create("BeginLocation", "System.String"),
            Tuple.Create("EndLocation", "System.String"),
            Tuple.Create("BeginLat", "System.Double"),
            Tuple.Create("BeginLon", "System.Double"),
            Tuple.Create("EndLat", "System.Double"),
            Tuple.Create("EndLon", "System.Double"),
            Tuple.Create("EpisodeNarrative", "System.String"),
            Tuple.Create("EventNarrative", "System.String"),
            Tuple.Create("StormSummary", "System.Object"),
        }
    );
    await kustoClient.ExecuteControlCommandAsync(databaseName, command);
}

Veri alımı eşlemesini tanımlama

Gelen CSV verilerini tabloyu oluştururken kullanılan sütun adlarına eşleyin. Bu tabloda bir CSV sütun eşleme nesnesi sağlayın.

var tableMappingName = "StormEvents_CSV_Mapping";
using (var kustoClient = KustoClientFactory.CreateCslAdminProvider(kustoConnectionStringBuilder))
{
    var command = CslCommandGenerator.GenerateTableMappingCreateCommand(
        IngestionMappingKind.Csv,
        tableName,
        tableMappingName,
        new ColumnMapping[]
        {
            new() { ColumnName = "StartTime", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "0" } } },
            new() { ColumnName = "EndTime", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "1" } } },
            new() { ColumnName = "EpisodeId", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "2" } } },
            new() { ColumnName = "EventId", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "3" } } },
            new() { ColumnName = "State", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "4" } } },
            new() { ColumnName = "EventType", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "5" } } },
            new() { ColumnName = "InjuriesDirect", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "6" } } },
            new() { ColumnName = "InjuriesIndirect", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "7" } } },
            new() { ColumnName = "DeathsDirect", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "8" } } },
            new() { ColumnName = "DeathsIndirect", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "9" } } },
            new() { ColumnName = "DamageProperty", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "10" } } },
            new() { ColumnName = "DamageCrops", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "11" } } },
            new() { ColumnName = "Source", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "12" } } },
            new() { ColumnName = "BeginLocation", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "13" } } },
            new() { ColumnName = "EndLocation", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "14" } } },
            new() { ColumnName = "BeginLat", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "15" } } },
            new() { ColumnName = "BeginLon", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "16" } } },
            new() { ColumnName = "EndLat", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "17" } } },
            new() { ColumnName = "EndLon", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "18" } } },
            new() { ColumnName = "EpisodeNarrative", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "19" } } },
            new() { ColumnName = "EventNarrative", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "20" } } },
            new() { ColumnName = "StormSummary", Properties = new Dictionary<string, string> { { MappingConsts.Ordinal, "21" } } }
        }
    );
    
    await kustoClient.ExecuteControlCommandAsync(databaseName, command);
}

Tablonuz için toplu işlem ilkesi tanımlama

Gelen verileri toplu işleme, veri parça boyutunu en iyi duruma getirerek alma toplu işlem ilkesi tarafından denetlenmektedir. İlkeyi alma toplu ilkesi yönetimi komutuyla değiştirin. Yavaş gelen verilerin gecikme süresini azaltmak için bu ilkeyi kullanın.

using (var kustoClient = KustoClientFactory.CreateCslAdminProvider(kustoConnectionStringBuilder))
{
    var command = CslCommandGenerator.GenerateTableAlterIngestionBatchingPolicyCommand(
        databaseName,
        tableName,
        new IngestionBatchingPolicy(
            maximumBatchingTimeSpan: TimeSpan.FromSeconds(10),
            maximumNumberOfItems: 100,
            maximumRawDataSizeMB: 1024
        )
    );
    kustoClient.ExecuteControlCommand(command);
}

Alınan veriler için bir Raw Data Size değer tanımlamanızı ve performansın iyileşip iyileşmediğini denetlerken boyutu artımlı olarak 250 MB'a azaltmanızı öneririz.

Özelliğini kullanarak Flush Immediately toplu işlemi atlayabilirsiniz, ancak düşük performansa neden olabileceğinden büyük ölçekli alımlar için bu önerilmez.

Veri alımı için bir iletiyi kuyruğa alma

Blob depolamadan veri çekmek ve verileri almak için iletiyi kuyruğa alın. Alma kümesine bir bağlantı kurulur ve bu uç noktayla çalışmak için başka bir istemci oluşturulur.

var ingestUri = "https://ingest-<clusterName>.<region>.kusto.windows.net";
var ingestConnectionStringBuilder = new KustoConnectionStringBuilder(ingestUri).WithAadUserPromptAuthentication(tenantId);
using var ingestClient = KustoIngestFactory.CreateQueuedIngestClient(ingestConnectionStringBuilder);
var properties = new KustoQueuedIngestionProperties(databaseName, tableName)
{
    Format = DataSourceFormat.csv,
    IngestionMapping = new IngestionMapping
    {
        IngestionMappingReference = tableMappingName,
        IngestionMappingKind = IngestionMappingKind.Csv
    },
    IgnoreFirstRecord = true
};
await ingestClient.IngestFromStorageAsync(blobPath, properties);

Verilerin tabloya alındığını doğrulama

Kuyruğa alınan alımın alımı zamanlaması ve verileri kümenize yüklemesi için beş ile on dakika arasında bekleyin. Ardından aşağıdaki kodu çalıştırarak StormEvents tablosundaki kayıtların sayısını alın.

using var cslQueryProvider = KustoClientFactory.CreateCslQueryProvider(kustoConnectionStringBuilder);
var query = $"{tableName} | count";
var results = cslQueryProvider.ExecuteQuery<long>(databaseName, query);
Console.WriteLine(results.Single());

Sorun giderme sorguları çalıştırma

https://dataexplorer.azure.com adresinde oturum açın ve kümenize bağlanın. Son dört saatte hiç veri alımı hatası olup olmadığını görmek için veritabanınızda aşağıdaki komutu çalıştırın. Çalıştırmadan önce veritabanı adını değiştirin.

.show ingestion failures
| where FailedOn > ago(4h) and Database == "<DatabaseName>"

Son dört saatteki tüm veri alım işlemlerinin durumunu görüntülemek için aşağıdaki komutu çalıştırın. Çalıştırmadan önce veritabanı adını değiştirin.

.show operations
| where StartedOn > ago(4h) and Database == "<DatabaseName>" and Operation == "DataIngestPull"
| summarize arg_max(LastUpdatedOn, *) by OperationId

Kaynakları temizleme

Diğer makalelerimizi izlemeyi planlıyorsanız, oluşturduğunuz kaynakları koruyun. Aksi takdirde, veritabanınızda aşağıdaki komutu çalıştırarak StormEvents tablosunu temizleyin.

.drop table StormEvents

İlgili içerik

• Sorgu yazma