Kusto Query Language（KQL）のカスタム列の追加、データ型変換と条件分岐による値の変換、情報抽出 (SQL の補足付き)

データ分析やデータ処理の過程でカスタム列を追加したい場合や、データの型を変換したい場合がよくあります。  
また、特定の条件に応じて異なる値を表示したい場合もあります。  
この記事では、Kusto Query Language (KQL)と SQL を使用した、データ型の変換と条件分岐による値の変換について例を紹介します。

## カスタム列の追加、データ型変換と条件分岐による値の変換

### カスタム列の追加とデータ型の変換 (string から double へ)

このクエリでは、文字列型の `HoursWorked` 列を `double` 型に変換し、`HoursWorkedDouble` という新しい列に追加します。

```kql
// サンプルデータ
datatable(EmployeeId:string, HoursWorked:string)
[
    "E01", "40",
    "E02", "35.5"
]
// HoursWorkedをdouble型に変換し、新しい列として追加
| extend HoursWorkedDouble = todouble(HoursWorked)
```

SQL では、`CAST`関数を使用して`HoursWorked`列を DOUBLE に変換し、`HoursWorkedDouble` という新しい列に追加します。

```sql
SELECT EmployeeId, CAST(HoursWorked AS DOUBLE) AS HoursWorkedDouble
FROM WorkHours
```

### データ型の変換 (int から string へ)

このクエリでは、`OrderId`列のデータ型を int から string に変換しています。

```kql
// サンプルデータ
datatable(OrderId:int, Quantity:int)
[
    1, 10,
    2, 20
]
// OrderIdをstring型に変換
| project OrderIdString = tostring(OrderId), Quantity
```

SQL では、`CAST`関数を使用して`OrderId`列のデータ型を int から VARCHAR に変換します。

```sql
SELECT
    CAST(OrderId AS VARCHAR) AS OrderIdString,
    Quantity
FROM Orders
```

### 条件に基づく値の変換

`case`関数を使用して、`Status`の値に応じて異なる文字列を`StatusDescription`列として追加します。

```kql
// サンプルデータ
datatable(Status:int)
[
    1,
    2,
    3
]
// Statusに基づいてStatusDescription列を作成
| extend StatusDescription = case(Status == 1, "Pending",
                                  Status == 2, "In Progress",
                                  Status == 3, "Completed",
                                  "Unknown")
```

SQL では、`CASE`文を使用して同様の条件分岐を行い、`Status`の値に基づいて異なる文字列を表示します。

```sql
SELECT
    CASE
        WHEN Status = 1 THEN 'Pending'
        WHEN Status = 2 THEN 'In Progress'
        WHEN Status = 3 THEN 'Completed'
        ELSE 'Unknown'
    END AS StatusDescription
FROM OrderStatus
```

### KQL での変数宣言

KQL における変数の宣言は、`let` ステートメントを使用して行います。

```kql
let timeOffset = 1h;
let discardEventId = 4688;
SecurityEvent
| where TimeGenerated > ago(timeOffset*2) and TimeGenerated < ago(timeOffset)
| where EventID != discardEventId
```

SQL における変数の宣言は、`DECLARE` ステートメントを使用して行います。

```sql
DECLARE @TimeOffset INT = 60; -- 分単位で1時間
DECLARE @DiscardEventId INT = 4688;

SELECT *
FROM SecurityEvent
WHERE
    TimeGenerated > DATEADD(MINUTE, -@TimeOffset*2, GETUTCDATE()) -- 現在時刻から2時間前
    AND TimeGenerated < DATEADD(MINUTE, -@TimeOffset, GETUTCDATE()) -- 現在時刻から1時間前
    AND EventID != @DiscardEventId;
```

### KQL での動的リストの宣言

KQL の例。

```kql
let suspiciousAccounts = datatable(account: string) [
  @"NA\timadmin",
  @"NT AUTHORITY\SYSTEM"
];
SecurityEvent
| where TimeGenerated > ago(1h)
| where Account in (suspiciousAccounts)
```

SQL の例。

```sql
WITH SuspiciousAccounts(account) AS (
  SELECT 'NA\timadmin' UNION ALL
  SELECT 'NT AUTHORITY\SYSTEM'
)

SELECT se.*
FROM SecurityEvent se
JOIN SuspiciousAccounts sa ON se.Account = sa.account
WHERE se.TimeGenerated > DATEADD(HOUR, -1, GETUTCDATE());
```

### KQL で文字列データを操作する

#### 例 1: ログデータからのユーザー ID 抽出

この例では、`extract`関数を使ってログエントリからユーザー ID を抽出しています。  
正規表現 `"UserID=([0-9]+);"` を使用し、第一のキャプチャグループに一致する部分（ユーザー ID）を選択しています。

```kql
let LogData = datatable(LogEntry:string)
[
    "User Logged In: UserID=12345; SessionID=abcde",
    "Error: Something went wrong",
    "User Logged In: UserID=67890; SessionID=fghij"
];
LogData
| project ExtractedUserID = extract("UserID=([0-9]+);", 1, LogEntry)
```

SQL では、正規表現のサポートはデータベースシステムによって異なります。以下は、PostgreSQL での例です。  
この SQL クエリは、PostgreSQL の`SUBSTRING`関数と正規表現を使って同様の操作を実行しています。

```sql
SELECT
    SUBSTRING(log_entry FROM 'UserID=([0-9]+);') AS extracted_user_id
FROM
    (VALUES
        ('User Logged In: UserID=12345; SessionID=abcde'),
        ('Error: Something went wrong'),
        ('User Logged In: UserID=67890; SessionID=fghij')
    ) AS log_data(log_entry);
```

#### 例 2: メールアドレスからドメイン名の抽出

メールアドレスのリストからドメイン名部分を抽出する操作を考えます。  
このクエリは、`extract`関数を使ってメールアドレスのドメイン名部分を抽出しています。`@(.+)$`という正規表現を用いています。

```kql
let Emails = datatable(EmailAddress:string)
[
    "user1@example.com",
    "user2@sample.org"
];
Emails
| project Domain = extract("@(.+)$", 1, EmailAddress)
```

ここでも、SQL の例として PostgreSQL を用います。  
この SQL クエリも、`SUBSTRING`関数と正規表現を使ってメールアドレスからドメイン部分を抽出しています。

```sql
SELECT
    SUBSTRING(email_address FROM '@(.+)$') AS domain
FROM
    (VALUES
        ('user1@example.com'),
        ('user2@sample.org')
    ) AS emails(email_address);
```

### KQL で JSON データから情報を抽出する

ログデータに JSON 形式で格納されている情報を効率的に抽出する手法は、分析や・モニタリング作業上重要です。  
このクエリでは`parse_json`関数を使用して JSON データを解析し、必要な情報を`project`ステートメントで選択しています。  
ログデータ内の`Details`列が JSON オブジェクトを含んでいるとします。この JSON から`userId`と`actionType`を抽出するクエリは以下の通りです。

```kql
datatable(EventLog:string)
[
    '{"userId": "u123", "actionType": "login", "time": "2020-01-01T12:00:00Z"}',
    '{"userId": "u124", "actionType": "logout", "time": "2020-01-01T12:05:00Z"}'
]
| extend DetailsJson = parse_json(EventLog)
| project UserId = DetailsJson.userId, ActionType = DetailsJson.actionType
```

SQL の場合、多くの現代のデータベースシステムは JSON データの処理機能をサポートしています。以下は、PostgreSQL で JSON データから情報を抽出する例です。  
この SQL クエリでは、PostgreSQL の JSON 演算子`->>`を使用して、文字列として JSON オブジェクトから値を取得しています。

```sql
SELECT
    json_data ->> 'userId' AS UserId,
    json_data ->> 'actionType' AS ActionType
FROM
    (SELECT '{"userId": "u123", "actionType": "login", "time": "2020-01-01T12:00:00Z"}'::json AS json_data) AS logs
```

## 参考記事

- [SQL から Kusto のチート シート](https://learn.microsoft.com/ja-jp/azure/data-explorer/kusto/query/sql-cheat-sheet)
- [Microsoft Learn](https://microsoftlearning.github.io/SC-200T00A-Microsoft-Security-Operations-Analyst.ja-jp/Instructions/Labs/LAB_AK_04_Lab1_Ex1_KQL.html)