JackyDataAnalysisLibrary.Connectors.Adx 0.1.0

JackyDataAnalysisLibrary

一個乾淨、低耦合的 C# 資料分析函式庫，將查詢語言建構層（KQL / T-SQL / PostgreSQL SQL）從應用程式中獨立抽出，讓多個異質資料來源可以透過同一套介面查詢，不需在應用層感知底層是哪種資料庫。

安裝

依需要的資料來源選擇安裝，Core 是必裝的基礎套件，Connector 則按需安裝：

# 必裝
dotnet add package JackyDataAnalysisLibrary.Core

# 擇一或多個 Connector
dotnet add package JackyDataAnalysisLibrary.Connectors.Adx        # Azure Data Explorer (KQL)
dotnet add package JackyDataAnalysisLibrary.Connectors.MsSql       # Microsoft SQL Server
dotnet add package JackyDataAnalysisLibrary.Connectors.PostgreSql  # PostgreSQL

快速開始

步驟一：建立引擎

用 DataAnalysisEngineBuilder 登錄資料來源，Build() 後得到引擎實例。引擎建好後是 thread-safe，可以 Singleton 注入。

// ADX（Service Principal 認證）
var adxConfig = new AdxConfig
{
    SourceKey    = "adx",
    ClusterUri   = "https://yourcluster.kusto.windows.net",
    Database     = "YourDatabase",
    ClientId     = "your-client-id",
    ClientSecret = "your-client-secret",
    TenantId     = "your-tenant-id"
};

// MSSQL
var mssqlConfig = new MsSqlConfig
{
    SourceKey        = "mssql",
    ConnectionString = "Server=myserver;Database=mydb;Integrated Security=true;"
};

// PostgreSQL
var pgConfig = new PostgreSqlConfig
{
    SourceKey        = "pg",
    ConnectionString = "Host=localhost;Database=mydb;Username=postgres;Password=secret"
};

var engine = new DataAnalysisEngineBuilder()
    .AddSource(new AdxDataSourceFactory(adxConfig))
    .AddSource(new MsSqlDataSourceFactory(mssqlConfig))
    .AddSource(new PostgreSqlDataSourceFactory(pgConfig))
    .Build();

ASP.NET Core DI 整合：

builder.Services.AddSingleton(engine);

步驟二：選擇查詢格式

引擎提供四種查詢格式，透過 SourceKey 指定目標資料來源。

四種查詢格式

Summarize — 聚合查詢

對應 SQL GROUP BY，ADX summarize，用於取得彙總指標。

var result = await engine.SummarizeAsync(new SummarizeRequest
{
    SourceKey = "mssql",
    TableName = "Orders",
    Filters   = [new FilterDimension("Region", ["APAC", "EMEA"])],
    ViewBy    = [new GroupField("Region"), new GroupField("ProductCategory")],
    Measures  =
    [
        new MeasureField("Revenue",      AggregationType.Sum),
        new MeasureField("OrderId",      AggregationType.Count,         Alias: "OrderCount"),
        new MeasureField("CustomerId",   AggregationType.DistinctCount, Alias: "UniqueCustomers"),
        new MeasureField("UnitPrice",    AggregationType.Average,       Alias: "AvgPrice"),
    ]
});

if (result.IsSuccess)
{
    foreach (var row in result.Value!.Rows)
    {
        var region   = row.Get<string>("Region");
        var revenue  = row.Get<decimal>("Revenue");
        var orders   = row.Get<int>("OrderCount");
        Console.WriteLine($"{region}: {revenue:C} ({orders} orders)");
    }
}

所有 AggregationType：

Details — 明細查詢

取得 row-level 原始資料，分頁由引擎下推至資料來源，不在應用記憶體做。

var result = await engine.DetailsAsync(new DetailsRequest
{
    SourceKey    = "adx",
    TableName    = "EventLogs",
    SelectFields = ["EventId", "Timestamp", "Level", "Message"],   // 空白 = 全欄位
    Filters      =
    [
        new FilterDimension("Level",  ["ERROR", "CRITICAL"]),
        new FilterDimension("Region", ["TW"], FilterOperator.NotIn),
    ],
    OrderBy = [new SortField("Timestamp", Descending: true)],
    Skip    = 0,
    Take    = 50,   // 每頁 50 筆
});

if (result.IsSuccess)
{
    Console.WriteLine($"本頁：{result.Value!.TotalCount} 筆");
    foreach (var row in result.Value.Rows)
        Console.WriteLine($"[{row["Level"]}] {row["Message"]}");
}

FieldList — Schema 查詢

取得資料表的欄位定義，含 native type 與跨來源統一的 CanonicalType，適合動態渲染 UI 元件。

var result = await engine.FieldListAsync(new FieldListRequest
{
    SourceKey = "pg",
    TableName = "customers"
});

if (result.IsSuccess)
{
    foreach (var field in result.Value!.Fields)
    {
        // DataType    = native type 字串，如 "character varying"、"integer"、"boolean"
        // CanonicalType = 跨來源統一 enum，如 String、Integer、Boolean
        Console.WriteLine($"{field.Name,-20} {field.DataType,-25} → {field.CanonicalType}  nullable={field.IsNullable}");
    }
}

輸出範例（PostgreSQL customers 表）：

id                   integer                   → Integer  nullable=False
name                 character varying         → String   nullable=False
email                character varying         → String   nullable=True
created_at           timestamp with time zone  → DateTimeOffset  nullable=False
is_active            boolean                   → Boolean  nullable=False

用 CanonicalType 動態決定 UI 元件

foreach (var field in result.Value!.Fields)
{
    var component = field.CanonicalType switch
    {
        ColumnDataType.DateTime or
        ColumnDataType.DateTimeOffset or
        ColumnDataType.Date            => "DatePicker",

        ColumnDataType.Boolean         => "Toggle",

        ColumnDataType.Integer  or
        ColumnDataType.Float    or
        ColumnDataType.Decimal         => "NumberInput",

        ColumnDataType.Json            => "JsonEditor",

        _                              => "TextInput"   // String、Unknown...
    };

    Console.WriteLine($"{field.Name} → {component}");
}

完整 ColumnDataType 列表：

MaxDate — 日期邊界查詢

取得指定日期欄位的最大值（MAX()）。語意上是「欄位的上界」，與資料是否含未來時間無關，常用於確認資料新鮮度或計算時間範圍。

var result = await engine.MaxDateAsync(new MaxDateRequest
{
    SourceKey     = "adx",
    TableName     = "TelemetryEvents",
    DateFieldName = "Timestamp",
    Filters       = [new FilterDimension("DeviceType", ["Sensor"])]
});

if (result.IsSuccess)
{
    var bound = result.Value!.BoundDate;
    if (bound is null)
        Console.WriteLine("資料表為空");
    else
        Console.WriteLine($"資料上界：{bound:yyyy-MM-dd HH:mm:ss zzz}");
}

篩選條件（FilterDimension）

// IN（預設）— 最常用
new FilterDimension("Status", ["Active", "Pending"])

// NOT IN
new FilterDimension("Region", ["TW"], FilterOperator.NotIn)

// 單值比較
new FilterDimension("Score",    ["90"], FilterOperator.GreaterThan)
new FilterDimension("Priority", ["3"],  FilterOperator.LessThan)
new FilterDimension("Code",     ["US"], FilterOperator.Equals)

// 範圍（Between 需傳兩個值：[min, max]）
new FilterDimension("Amount", ["100", "500"], FilterOperator.Between)

// 模糊搜尋（Contains 轉換為 LIKE / ILIKE）
new FilterDimension("Name", ["John"], FilterOperator.Contains)

ADX 限制： ADX 僅支援 In / NotIn，其他運算子會拋出 NotSupportedException。

Result Pattern

所有查詢方法均回傳 QueryResult<T>，不會拋出例外（OperationCanceledException 除外）。

var result = await engine.SummarizeAsync(request);

if (result.IsSuccess)
{
    var data = result.Value!;  // 保證非 null
    // ... 使用 data
}
else
{
    // 記錄錯誤或回傳給呼叫端
    logger.LogError("Query failed [{Code}]: {Message}", result.ErrorCode, result.ErrorMessage);
}

內建 ErrorCode：

CancellationToken 支援

所有查詢方法均接受 CancellationToken，適合 Web API 的 request abort 場景：

// ASP.NET Core Controller
public async Task<IActionResult> GetSummary(CancellationToken ct)
{
    var result = await engine.SummarizeAsync(request, ct);
    // 若 request 被 client 取消，ExecuteAsync 會拋出 OperationCanceledException
    // ASP.NET Core 會自動處理，不需在此 catch
    ...
}

擴充新資料來源

只需新增套件，不修改任何現有程式碼（OCP）：

var engine = new DataAnalysisEngineBuilder()
    .AddSource(new AdxDataSourceFactory(adxConfig))
    .AddSource(new MyCustomDataSourceFactory(myConfig))  // ← 新增這行
    .Build();

詳細實作步驟請參閱 Developer Guide。

開發環境需求

• .NET 8.0+
• C# 12+