Kusto.Ingest クライアント インターフェイスとクラス
Kusto Ingest .NET ライブラリのメイン インターフェイスとクラスは次のとおりです。
- インターフェイス IKustoIngestClient: 主なインジェスト インターフェイス
- クラス ExtendedKustoIngestClient: 主なインジェスト インターフェイスの拡張機能
- クラス KustoIngestFactory: インジェストクライアントの主なファクトリ
- クラス KustoIngestionProperties: 一般的なインジェスト プロパティを提供するために使用されるクラス
- クラス SourceOptions: ソースデータ処理オプション
- クラス IngestionMapping: インジェストのデータマッピングを説明するために使用されるクラス
- Enum DataSourceFormat: サポートされているデータ ソース形式。 たとえば、CSV や JSON などです。
- インターフェイス IKustoQueuedIngestClient: キューに入っているインジェストにのみ適用される操作を説明するインターフェイス
- クラス KustoQueuedIngestionProperties: キューに入っているインジェストにのみ適用されるプロパティ
注意
この情報は、特に .NET クライアント ライブラリに適用されます。 クライアント ライブラリ によって、使用可能なインターフェイスとクラスが異なる場合があります。 すべてのクライアント ライブラリにわたる Kusto Ingest の詳細については、「 Kusto Ingest の概要」を参照してください。
インターフェイス IKustoIngestClient
- IngestFromDataReaderAsync
- IngestFromStorageAsync
- IngestFromStreamAsync
public interface IKustoIngestClient : IDisposable
{
/// <summary>
/// Ingests data from <see cref="IDataReader"/>. <paramref name="dataReader"/> will be closed when the call completes.
/// </summary>
/// <param name="dataReader">The <see cref="IDataReader"/> data source to ingest. Only the first record set will be used</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="sourceOptions">Options for the <see cref="IDataReader"/> ingestion source. This is an optional parameter</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
Task<IKustoIngestionResult> IngestFromDataReaderAsync(IDataReader dataReader, KustoIngestionProperties ingestionProperties, DataReaderSourceOptions sourceOptions = null);
/// <summary>
/// Ingest data from one of the supported storage providers. Currently the supported providers are: File System, Azure Blob Storage.
/// </summary>
/// <param name="uri">The URI of the storage resource to be ingested. Note: This URI may include a storage account key or shared access signature (SAS).
/// See <see href="https://learn.microsoft.com/azure/kusto/api/connection-strings/storage"/> for the URI format options.</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="sourceOptions">Options for the storage ingestion source. This is an optional parameter</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
Task<IKustoIngestionResult> IngestFromStorageAsync(string uri, KustoIngestionProperties ingestionProperties, StorageSourceOptions sourceOptions = null);
/// <summary>
/// Ingest data from <see cref="Stream"/>.
/// </summary>
/// <param name="stream">The <see cref="Stream"/> data source to ingest</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="sourceOptions">Options for the <see cref="Stream"/> ingestion source. This is an optional parameter</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
Task<IKustoIngestionResult> IngestFromStreamAsync(Stream stream, KustoIngestionProperties ingestionProperties, StreamSourceOptions sourceOptions = null);
}
クラス ExtendedKustoIngestClient
- IngestFromSingleBlob - 非推奨。 代わりに
IKustoIngestClient.IngestFromStorageAsync
を使用してください - IngestFromSingleBlobAsync - 非推奨。 代わりに
IKustoIngestClient.IngestFromStorageAsync
を使用してください - IngestFromDataReader - 非推奨。 代わりに
IKustoIngestClient.IngestFromDataReaderAsync
を使用してください - IngestFromDataReaderAsync
- IngestFromSingleFile - 非推奨。 代わりに
IKustoIngestClient.IngestFromStorageAsync
を使用してください - IngestFromSingleFileAsync - 非推奨。 代わりに
IKustoIngestClient.IngestFromStorageAsync
を使用してください - IngestFromStream - 非推奨。 代わりに
IKustoIngestClient.IngestFromStreamAsync
を使用してください - IngestFromStreamAsync
public static class ExtendedKustoIngestClient
{
/// <summary>
/// Ingest data from a single data blob
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="blobUri">The URI of the blob will be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source blob should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="rawDataSize">The uncompressed raw data size</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromSingleBlob(this IKustoIngestClient client, string blobUri, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties, long? rawDataSize = null);
/// <summary>
/// Ingest data from a single data blob asynchronously
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="blobUri">The URI of the blob will be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source blob should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="rawDataSize">The uncompressed raw data size</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromSingleBlobAsync(this IKustoIngestClient client, string blobUri, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties, long? rawDataSize = null);
/// <summary>
/// Ingest data from a single data blob
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="blobDescription"><see cref="BlobDescription"/> representing the blobs that will be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source blob should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="rawDataSize">The uncompressed raw data size</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromSingleBlob(this IKustoIngestClient client, BlobDescription blobDescription, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties, long? rawDataSize = null);
/// <summary>
/// Ingest data from a single data blob asynchronously
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="blobDescription"><see cref="BlobDescription"/> representing the blobs that will be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source blob should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="rawDataSize">The uncompressed raw data size</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromSingleBlobAsync(this IKustoIngestClient client, BlobDescription blobDescription, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties, long? rawDataSize = null);
/// <summary>
/// Ingest data from <see cref="IDataReader"/>, which is closed and disposed of upon call completion
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="dataReader">The data to ingest (only the first record set will be used)</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromDataReader(this IKustoIngestClient client, IDataReader dataReader, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Asynchronously ingest data from <see cref="IDataReader"/>, which is closed and disposed of upon call completion
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="dataReader">The data to ingest (only the first record set will be used)</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromDataReaderAsync(this IKustoIngestClient client, IDataReader dataReader, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Ingest data from <see cref="IDataReader"/>, which is closed and disposed of upon call completion
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="dataReaderDescription"><see cref="DataReaderDescription"/>Represents the data to ingest (only the first record set will be used)</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromDataReader(this IKustoIngestClient client, DataReaderDescription dataReaderDescription, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Asynchronously ingest data from <see cref="IDataReader"/>, which is closed and disposed of upon call completion
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="dataReaderDescription"><see cref="DataReaderDescription"/>Represents the data to ingest (only the first record set will be used)</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromDataReaderAsync(this IKustoIngestClient client, DataReaderDescription dataReaderDescription, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Ingest data from a single file
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="filePath">Absolute path of the source file to be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source file should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromSingleFile(this IKustoIngestClient client, string filePath, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Ingest data from a single file asynchronously
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="filePath">Absolute path of the source file to be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source file should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromSingleFileAsync(this IKustoIngestClient client, string filePath, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Ingest data from a single file
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="fileDescription"><see cref="FileDescription"/> representing the file that will be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source file should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromSingleFile(this IKustoIngestClient client, FileDescription fileDescription, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Ingest data from a single file asynchronously
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="fileDescription"><see cref="FileDescription"/> representing the file that will be ingested</param>
/// <param name="deleteSourceOnSuccess">Indicates if the source file should be deleted after a successful ingestion</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromSingleFileAsync(this IKustoIngestClient client, FileDescription fileDescription, bool deleteSourceOnSuccess, KustoIngestionProperties ingestionProperties);
/// <summary>
/// Ingest data from <see cref="Stream"/>
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="stream">The data to ingest</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="leaveOpen">Optional. If set to 'false' (default value), <paramref name="stream"/> will be closed and disposed on call completion</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromStream(this IKustoIngestClient client, Stream stream, KustoIngestionProperties ingestionProperties, bool leaveOpen = false);
/// <summary>
/// Ingest data from <see cref="Stream"/> asynchronously
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="stream">The data to ingest</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="leaveOpen">Optional. If set to 'false' (default value), <paramref name="stream"/> will be closed and disposed on call completion</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromStreamAsync(this IKustoIngestClient client, Stream stream, KustoIngestionProperties ingestionProperties, bool leaveOpen = false);
/// <summary>
/// Ingest data from <see cref="Stream"/>
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="streamDescription"><see cref="StreamDescription"/>Represents the data to ingest</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="leaveOpen">Optional. If set to 'false' (default value), streamDescription.Stream will be closed and disposed on call completion</param>
/// <returns><see cref="IKustoIngestionResult"/></returns>
public static IKustoIngestionResult IngestFromStream(this IKustoIngestClient client, StreamDescription streamDescription, KustoIngestionProperties ingestionProperties, bool leaveOpen = false);
/// <summary>
/// Ingest data from <see cref="Stream"/> asynchronously
/// </summary>
/// <param name="client">The ingest client that will execute the ingestions</param>
/// <param name="streamDescription"><see cref="StreamDescription"/>Represents the data to ingest</param>
/// <param name="ingestionProperties">Additional properties to be used during the ingestion process</param>
/// <param name="leaveOpen">Optional. If set to 'false' (default value), streamDescription.Stream will be closed and disposed on call completion</param>
/// <returns>An <see cref="IKustoIngestionResult"/> task</returns>
public static Task<IKustoIngestionResult> IngestFromStreamAsync(this IKustoIngestClient client, StreamDescription streamDescription, KustoIngestionProperties ingestionProperties, bool leaveOpen = false);
}
クラス KustoIngestFactory
- CreateDirectIngestClient
- CreateQueuedIngestClient
- CreateManagedStreamingIngestClient
- CreateStreamingIngestClient
/// <summary>
/// Factory for creating Kusto ingestion objects.
/// </summary>
public static class KustoIngestFactory
{
/// <summary>
/// Creates an implementation of <see cref="IKustoIngestClient"/> that communicates
/// directly with the Kusto engine service.
/// </summary>
/// <param name="kcsb">Indicates the connection to the Kusto engine service.</param>
/// <returns>An implementation of <see cref="IKustoIngestClient"/> that communicates
/// directly with the Kusto engine service.</returns>
/// <remarks>In most cases, it is preferred that ingestion be done using the
/// queued implementation of <see cref="IKustoIngestClient"/>. See <see cref="CreateQueuedIngestClient(KustoConnectionStringBuilder)"/>.</remarks>
public static IKustoIngestClient CreateDirectIngestClient(KustoConnectionStringBuilder kcsb);
/// <summary>
/// Creates an implementation of <see cref="IKustoIngestClient"/> that communicates
/// directly with the Kusto engine service.
/// </summary>
/// <param name="connectionString">Indicates the connection to the Kusto engine service.</param>
/// <returns>An implementation of <see cref="IKustoIngestClient"/> that communicates
/// directly with the Kusto engine service.</returns>
/// <remarks>In most cases, it is preferred that ingestion be done using the
/// queued implementation of <see cref="IKustoIngestClient"/>. See <see cref="CreateQueuedIngestClient(string)"/>.</remarks>
public static IKustoIngestClient CreateDirectIngestClient(string connectionString);
/// <summary>
/// Creates an implementation of <see cref="IKustoQueuedIngestClient"/> that communicates
/// with the Kusto ingestion service using a reliable queue.
/// </summary>
/// <param name="kcsb">Indicates the connection to the Kusto ingestion service.
/// Note that the ingestion service generally has a "ingest-" prefix in the
/// DNS host name part.</param>
/// <returns>An implementation of <see cref="IKustoQueuedIngestClient"/> that communicates
/// with the Kusto ingestion service using a reliable queue.</returns>
public static IKustoQueuedIngestClient CreateQueuedIngestClient(KustoConnectionStringBuilder kcsb);
/// <summary>
/// Creates an implementation of <see cref="IKustoQueuedIngestClient"/> that communicates
/// with the Kusto ingestion service using a reliable queue.
/// </summary>
/// <param name="connectionString">Indicates the connection to the Kusto ingestion service.
/// Note that the ingestion service generally has a "ingest-" prefix in the
/// DNS host name part.</param>
/// <returns>An implementation of <see cref="IKustoQueuedIngestClient"/> that communicates with the Kusto ingestion service using a reliable queue.</returns>
public static IKustoQueuedIngestClient CreateQueuedIngestClient(string connectionString);
/// <summary>
/// Creates an implementation of <see cref="IKustoIngestClient"/> that performs managed streaming ingestion
/// </summary>
/// <param name="engineKcsb">Indicates the connection to the Kusto engine service.</param>
/// <param name="dmKcsb">Indicates the connection to the Kusto data management service.</param>
/// <returns>An implementation of <see cref="IKustoIngestClient"/> that performs managed streaming ingestion</returns>
/// <remarks>Streaming ingestion is performed directy intto Kusto enginge cluster
/// and is optimized for low-latency ingestion of relatively small chunks of data.
/// If the streaming ingset doesn't succeed after several retries, queued ingestion will be performed.</remarks>
public static IKustoIngestClient CreateManagedStreamingIngestClient(KustoConnectionStringBuilder engineKcsb, KustoConnectionStringBuilder dmKcsb);
/// <summary>
/// Creates an implementation of <see cref="IKustoIngestClient"/> that performs managed streaming ingestion
/// </summary>
/// <param name="engineConnectionString">Indicates the connection to the Kusto engine service.</param>
/// <param name="dmConnectionString">Indicates the connection to the Kusto data management service.</param>
/// <returns>An implementation of <see cref="IKustoIngestClient"/> that performs managed streaming ingestion</returns>
/// <remarks>Streaming ingestion is performed directy intto Kusto enginge cluster
/// and is optimized for low-latency ingestion of relatively small chunks of data.
/// If the streaming ingset doesn't succeed after several retries, queued ingestion will be performed.</remarks>
public static IKustoIngestClient CreateManagedStreamingIngestClient(string engineConnectionString, string dmConnectionString);
/// <summary>
/// Creates an implementation of <see cref="IKustoIngestClient"/> that performs streaming ingestion
/// </summary>
/// <param name="kcsb">Indicates the connection to the Kusto engine service.</param>
/// <returns>An implementation of <see cref="IKustoIngestClient"/> that performs streaming ingestion</returns>
/// <remarks>Streaming ingestion is performed directy intto Kusto enginge cluster
/// and is optimized for low-latency ingestion of relatively small chunks of data</remarks>
public static IKustoIngestClient CreateStreamingIngestClient(KustoConnectionStringBuilder kcsb);
/// <summary>
/// Creates an implementation of <see cref="IKustoIngestClient"/> that performs streaming ingestion
/// </summary>
/// <param name="connectionString">Indicates the connection to the Kusto engine service.</param>
/// <returns>An implementation of <see cref="IKustoIngestClient"/> that performs streaming ingestion</returns>
/// <remarks>Streaming ingestion is performed directy into Kusto enginge cluster
/// and is optimized for low-latency ingestion of relatively small chunks of data</remarks>
public static IKustoIngestClient CreateStreamingIngestClient(string connectionString);
}
クラス KustoIngestionProperties
KustoIngestionProperties クラスには、インジェスト プロセスを細かく制御するための基本的なインジェスト プロパティが含まれています。
プロパティ | 説明 |
---|---|
DatabaseName | 取り込むデータベースの名前 |
TableName | 取り込むテーブルの名前 |
DropByTags | 各エクステントに含まれるタグ。 DropByTags は永続的であり、.show table T extents where tags has 'some tag' または .drop extents <| .show table T extents where tags has 'some tag' のように使用できます |
IngestByTags | エクステントごとに記述されるタグ。 後で プロパティと共に IngestIfNotExists 使用して、同じデータを 2 回取り込むのを避けることができます。 詳細については、ingest-by: タグに関する記事を参照してください。 |
IngestionMapping | 終了したマッピングへの参照または列マッピングの一覧を保持します |
AdditionalTags | 必要に応じて追加のタグ |
IngestIfNotExists | 再度取り込みたくないタグの一覧 (テーブルごと) |
ValidationPolicy | データ検証の定義。 |
フォーマット | 取り込まれているデータの形式 |
AdditionalProperties | インジェスト プロパティとしてインジェスト コマンドに渡されるその他のプロパティ。 すべてのインジェスト プロパティがこのクラスの別のメンバーで表されるわけではないため、プロパティが渡されます |
public class KustoIngestionProperties
{
public string DatabaseName { get; set; }
public string TableName { get; set; }
public IEnumerable<string> DropByTags { get; set; }
public IEnumerable<string> IngestByTags { get; set; }
public IEnumerable<string> AdditionalTags { get; set; }
public IEnumerable<string> IngestIfNotExists { get; set; }
public IngestionMapping IngestionMapping { get; set; }
public ValidationPolicy ValidationPolicy { get; set; }
public DataSourceFormat? Format { get; set; }
public bool IgnoreSizeLimit { get; set; } // Determines whether the limit of 4GB per single ingestion source should be ignored. Defaults to false.
public IDictionary<string, string> AdditionalProperties { get; set; }
public KustoIngestionProperties(string databaseName, string tableName);
}
クラス SourceOptions
SourceOptions および派生クラスは、ソース データの追加情報と処理オプションをカプセル化します。 サポートされているソースによって仕様が異なります
// Base class
public abstract class SourceOptions
{
public Guid SourceId { get; set; }; // Identifies the ingestion source
public bool Compress { get; set; }; // Determines whether data should be compressed before being uploaded.
}
// Represents a local file/blob/ADLSv2 file
public sealed class StorageSourceOptions : SourceOptions
{
public long? Size { get; set; }; // Uncompressed data size. Should be used to comunicate the file size to the service for efficient ingestion
public bool DeleteSourceOnSuccess { get; set; }; // Indicates whether the ingestion source should be deleted after successful ingestion. Defaults to 'false'. When set to 'true', will require the service to individually delete each blob, which could put extra pressure on the service.
public DataSourceCompressionType CompressionType { get; set; }; // Indicates compression used. Defaults to 'none'
}
クラス IngestionMapping
既存のマッピングへの参照または列マッピングの一覧を保持します。
プロパティ | 説明 |
---|---|
IngestionMappings | 各列にターゲット列のデータとそのソースが説明されている列マッピング |
IngestionMappingKind | IngestionMappings プロパティに説明されているマッピングの種類: Csv、Json、Avro、Parquet、SStream、Orc、ApacheAvro、または W3CLogFileのうちの 1 つ |
IngestionMappingReference | 事前に作成されたマッピング名 |
public class IngestionMapping
{
public IEnumerable<ColumnMapping> IngestionMappings { get; set; }
public IngestionMappingKind IngestionMappingKind { get; set; }
public string IngestionMappingReference { get; set; }
public IngestionMapping()
public IngestionMapping(IngestionMapping ingestionMapping)
}
列挙型 DataSourceFormat
public enum DataSourceFormat
{
csv, // Data is in a CSV(-comma-separated values) format
tsv, // Data is in a TSV(-tab-separated values) format
scsv, // Data is in a SCSV(-semicolon-separated values) format
sohsv, // Data is in a SOHSV(-SOH (ASCII 1) separated values) format
psv, // Data is in a PSV (pipe-separated values) format
tsve, // Tab-separated value with '\' escaping character.
txt, // Each record is a line and has just one field
raw, // The entire stream/file/blob is a single record having a single field
json, // Data is in a JSON-line format (each line is record with a single JSON value)
multijson, // The data stream is a concatenation of JSON documents (property bags all)
avro, // Data is in a AVRO format
orc, // Data is in a ORC format
parquet, // Data is in a Parquet format
}
KustoIngestionProperties 定義の例
var guid = new Guid().ToString();
var kustoIngestionProperties = new KustoIngestionProperties("TargetDatabase", "TargetTable")
{
DropByTags = new List<string> { DateTime.Today.ToString() },
IngestByTags = new List<string> { guid },
AdditionalTags = new List<string> { "some tags" },
IngestIfNotExists = new List<string> { guid },
IngestionMapping = new IngestionMapping() {
IngestionMappingKind = Data.Ingestion.IngestionMappingKind.Csv,
IngestionMappings = new ColumnMapping[] { new ColumnMapping() {
ColumnName = "stringColumn",
Properties = new Dictionary<string, string>() {
{ MappingConsts.Ordinal, "1"} }
} },
},
ValidationPolicy = new ValidationPolicy { ValidationImplications = ValidationImplications.Fail, ValidationOptions = ValidationOptions.ValidateCsvInputConstantColumns },
Format = DataSourceFormat.csv
};
インターフェイス IKustoQueuedIngestClient
IKustoQueuedIngestClient インターフェイスは、インジェスト操作の結果に従い、インジェスト クライアントの RetryPolicy を公開する追跡メソッドを追加します。
- PeekTopIngestionFailures
- GetAndDiscardTopIngestionFailures
- GetAndDiscardTopIngestionSuccesses
public interface IKustoQueuedIngestClient : IKustoIngestClient
{
/// <summary>
/// Peeks top (== oldest) ingestion failures
/// </summary>
/// <param name="messagesLimit">Maximum ingestion failures to peek. Default value peeks 32 messages.</param>
/// <returns>A task which its result contains IEnumerable of <see cref="IngestionFailure"/>. The received messages won't be discarded from the relevant azure queue.</returns>
Task<IEnumerable<IngestionFailure>> PeekTopIngestionFailures(int messagesLimit = -1);
/// <summary>
/// Returns and deletes top (== oldest) ingestion failure notifications
/// </summary>
/// <param name="messagesLimit">Maximum ingestion failure notifications to get. Default value peeks 32 messages.</param>
/// <returns>A task which its result contains IEnumerable of <see cref="IngestionFailure"/>. The received messages will be discarded from the relevant azure queue.</returns>
Task<IEnumerable<IngestionFailure>> GetAndDiscardTopIngestionFailures(int messagesLimit = -1);
/// <summary>
/// Returns and deletes top (== oldest) ingestion success notifications
/// </summary>
/// <param name="messagesLimit">Maximum ingestion success notifications to get. Default value peeks 32 messages.</param>
/// <returns>A task which its result contains IEnumerable of <see cref="IngestionSuccess"/>. The received messages will be discarded from the relevant azure queue.</returns>
Task<IEnumerable<IngestionSuccess>> GetAndDiscardTopIngestionSuccesses(int messagesLimit = -1);
/// <summary>
/// An implementation of IRetryPolicy that will be enforced on every ingest call,
/// which affects how the ingest client handles retrying on transient failures
/// </summary>
IRetryPolicy QueueRetryPolicy { get; set; }
}
クラス KustoQueuedIngestionProperties
KustoQueuedIngestionProperties クラスは、インジェスト動作を微調整するために使用できるいくつかのコントロールのノブを持つ KustoIngestionProperties を拡張します。
プロパティ | 説明 |
---|---|
FlushImmediately | 既定値は false です。 true に設定すると、データ管理サービスの集計メカニズムがバイパスされます |
IngestionReportLevel | インジェスト ステータス レポートのレベルを制御します (既定値: FailuresOnly )。 最適なパフォーマンスとストレージの使用のため、IngestionReportLevel を FailuresAndSuccesses に設定しないことをお勧めします |
IngestionReportMethod | インジェスト ステータス レポートのターゲットを制御します。 使用可能なオプションは、Azure キュー、Azure テーブル、またはその両方です。 既定値は Queue です。 |
public class KustoQueuedIngestionProperties : KustoIngestionProperties
{
/// <summary>
/// Allows to stop the batching phase and will cause to an immediate ingestion.
/// Defaults to 'false'.
/// </summary>
public bool FlushImmediately { get; set; }
/// <summary>
/// Controls the ingestion status report level.
/// Defaults to 'FailuresOnly'.
/// </summary>
public IngestionReportLevel ReportLevel { get; set; }
/// <summary>
/// Controls the target of the ingestion status reporting. Available options are Azure Queue, Azure Table, or both.
/// Defaults to 'Queue'.
/// </summary>
public IngestionReportMethod ReportMethod { get; set; }
/// <summary>
/// Controls the target of the ingestion status reporting. Available options are Azure Queue, Azure Table, or both.
/// Defaults to 'Queue'.
/// </summary>
public IngestionReportMethod ReportMethod;
public KustoQueuedIngestionProperties(string databaseName, string tableName);
}
フィードバック
https://aka.ms/ContentUserFeedback」を参照してください。
以下は間もなく提供いたします。2024 年を通じて、コンテンツのフィードバック メカニズムとして GitHub の issue を段階的に廃止し、新しいフィードバック システムに置き換えます。 詳細については、「フィードバックの送信と表示