PRQL クエリリファレンス

CSIRT-Pro は PRQL (Pipelined Relational Query Language) を採用しています。 PRQL で記述したクエリは、分析データストアが解釈できる問い合わせに自動変換されて実行されます。 パイプライン形式で、処理が上から下へ順に流れる書き方をします。

本ページの例は実機で検証しています

掲載した構文・関数・クエリ例は、稼働中の検索エンジンで実際に実行し、動作を確認したものです。 一部の結果テーブルは実データに対する実行結果（例）です。

---

クイックリファレンス（チートシート）

句（パイプラインの構成要素）

句	役割	例
from	検索対象テーブル	from \firewall_logs``
filter	行の絞り込み	filter action == "DENY"
derive	計算フィールドの追加	derive {kb = bytes / 1024}
select	フィールドの選択・別名	select {__time, src_ip}
group / aggregate	グループ集計	group {src_ip} (aggregate {cnt = count this})
sort	並べ替え（- で降順）	sort {-__time}
take	件数制限	take 100
join	テーブル結合	join \indicators` (this.src_ip == that.value)`
window	移動平均・累積・順位	window rows:-2..0 (derive {ma = average value})

演算子

演算子	意味	例
== != > < >= <=	比較	filter status >= 400
~=	正規表現マッチ	filter message ~= "error\|fail"
&& / \|\|	AND / OR（filter 内）	filter (a == 1 \|\| b == 2)
+ - * /	算術	derive {kb = bytes / 1024}
??	NULL の既定値（coalesce）	derive {u = user ?? "unknown"}
== null / != null	NULL 判定	filter src_ip != null

よく使う関数

用途	関数	例
件数	count this	aggregate {n = count this}
ユニーク数	count_distinct <col>	aggregate {u = count_distinct src_ip}
合計 / 平均	sum / average	aggregate {t = sum bytes}
最小 / 最大	min / max	aggregate {last = max __time}
相対時刻	(now) - (to_interval_day n)	filter __time > (now) - (to_interval_day 7)
時間バケット	to_start_of_hour <col>	group {h = to_start_of_hour __time} (...)
Map 取得	get_map <col> "key"	derive {sev = get_map tags "severity"}
条件分岐	cond_if <cond> <a> <b>	sum (cond_if (x == "y") 1 0)
正規表現の意味	~= は部分一致	^ $ で位置を固定

---

基本構文

from `<pipeline名>`
filter <条件>
select {<フィールド>}
group {<グループ化>} (aggregate {<集計>})
sort {<ソート>}
take <件数>

テーブル名

from に指定するテーブル名は Pipeline 名と同じです（バッククォートで囲みます）。 組織を識別するプレフィックスは自動的に付与されるため、ユーザーが指定する必要はありません。 ケースや脅威情報などのシステムテーブルも同じ書き方で検索できます。

---

from — テーブルの指定

from `firewall_logs`

Pipeline 名をバッククォートで囲んで指定します。 結合する場合は join を使います。

---

filter — 条件フィルタ

比較演算子

演算子	説明	例
==	等価	filter status == 200
!=	不等価	filter status != 404
>	より大きい	filter bytes > 1000
<	より小さい	filter response_time < 500
>=	以上	filter status >= 400
<=	以下	filter status <= 499
~=	正規表現マッチ（部分一致）	filter message ~= "error\|fail"

~= は正規表現の部分一致

~= の右辺は正規表現で、文字列のどこかにマッチすれば真になります。 先頭・末尾を固定するには ^ / $ を使います（例: ~= "^192\\.168\\."）。 先頭に (?i) を付けると大文字小文字を無視します（例: ~= "(?i)error"）。 PRQL の "..." 文字列はバックスラッシュをエスケープ処理するため、. などのメタ文字をリテラルにするには \\. と二重に書きます。

論理演算子

# AND（複数の filter を連結）
from `firewall_logs`
filter action == "DENY"
filter src_ip == "192.168.1.100"

# OR（|| を 1 つの filter 内で使用）
from `auth_logs`
filter (status == "failed" || status == "locked")

時間範囲フィルタ

# 絶対日時で指定
from `firewall_logs`
filter __time > @2025-01-01
filter __time < @2025-01-31

# 相対時間で指定（直近 24 時間）
from `firewall_logs`
filter __time > (now) - (to_interval_hour 24)

__time フィールド

__time はすべてのテーブルに存在するタイムスタンプフィールドです。 UI の時間範囲セレクタで指定した場合は、対応する filter が自動的に追加されます。 定期実行する検知ルールでは、絶対日時ではなく相対時間（(now) - (to_interval_*)）を使ってください。

NULL チェックと既定値

from `firewall_logs`
filter src_ip != null

# NULL のときに既定値を与える（coalesce）
from `messages`
derive {owner = user ?? "unknown"}

---

select — フィールド選択

from `firewall_logs`
select {__time, src_ip, dst_ip, action, bytes}

エイリアス（別名）

from `firewall_logs`
select {
  timestamp = __time,
  source = src_ip,
  destination = dst_ip,
  transferred = bytes
}

---

derive — 計算フィールド

from `web_access`
derive {
  response_kb = bytes / 1024,
  is_error = status >= 400
}

---

group / aggregate — 集計

基本集計

from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {cnt = count this}
)
sort {-cnt}
take 10

集計関数

関数	説明	例
count this	件数	n = count this
count_distinct <col>	ユニーク件数	u = count_distinct src_ip
sum <col>	合計（数値型のみ）	total = sum bytes
average <col>	平均（数値型のみ）	avg_time = average response_time
min <col>	最小値	first = min __time
max <col>	最大値	last = max __time
group_array <col>	値を配列に集約	ips = group_array src_ip
group_uniq_array <col>	ユニーク値を配列に集約	uniq_ips = group_uniq_array src_ip

sum / average は数値型のみ

sum と average は数値型のカラムにのみ使えます。 日時型（例: created_at、__time）には min / max は使えますが、sum / average は使えません（型エラーになります）。

複数キーでのグループ化

from `firewall_logs`
group {src_ip, action} (
  aggregate {cnt = count this}
)
sort {-cnt}

条件付き件数（慣用パターン）

count (条件) は条件で件数を絞り込めません（行数を数えるだけです）。 条件に一致した件数は sum (cond_if ...) で表します。

from `auth_logs`
group {src_ip} (
  aggregate {
    failed = sum (cond_if (action == "login_failed") 1 0),
    success = sum (cond_if (action == "login_success") 1 0)
  }
)
filter failed >= 5

時間バケット集計

from `firewall_logs`
group {hour = to_start_of_hour __time} (
  aggregate {cnt = count this}
)
sort {hour}

大規模集計の所要時間

生ログ全体を走査する集計は、対象が大きいほど時間がかかります。 検索時に生ログからフィールドを抽出する schema-on-read 方式のためです。 時間範囲を絞る、対象 Pipeline を限定する、filter を先に置くことで走査量を減らせます。

---

sort — ソート

# 降順（- プレフィックス）
from `firewall_logs`
sort {-__time}

# 昇順
from `firewall_logs`
sort {bytes}

# 複数キー
from `firewall_logs`
sort {-cnt, src_ip}

---

take — 件数制限

from `firewall_logs`
take 100

---

join — テーブル結合

from `firewall_logs`
join side:left `auth_logs` (==src_ip)
select {firewall_logs.__time, src_ip, auth_logs.username, action}

結合タイプ	説明
join	INNER JOIN
join side:left	LEFT JOIN
join side:right	RIGHT JOIN
join side:full	FULL OUTER JOIN

結合条件の書き方

両側で同じカラム名を突き合わせる場合は、(==カラム名) の短縮形が使えます。

join `auth_logs` (==src_ip)

異なるカラム名を突き合わせる場合は、結合元を this、結合先を that として明示します。

from `firewall_logs`
join `indicators` (this.src_ip == that.value)

---

window — ウィンドウ計算（移動平均・累積・順位）

window は、行を並べた状態で「直前 N 行」や「先頭からの累積」といった範囲を指定し、その範囲に対する集計や順位付けを行います。 SQL のウィンドウ関数（OVER (...)）に相当します。 並び順は直前の sort で決まり、group の中に置くとグループ（パーティション）ごとに計算します。

範囲の指定

指定	説明
rows:-2..0	現在行を含む直前 2 行（移動範囲）
rolling:3	直近 3 行の移動窓
expanding:true	先頭から現在行までの累積

累積（expanding）

from `firewall_logs`
sort {__time}
derive {one = 1}
window expanding:true (
  derive {running_total = sum one}
)

移動範囲（rows / rolling）

from `metrics`
sort {__time}
window rows:-2..0 (
  derive {moving_avg = average value}
)

順位付け（row_number / lag）

ウィンドウ内で使える主な関数:

関数	説明
row_number this	パーティション内の連番（1, 2, 3, …）
lag <n> <col> / lead <n> <col>	n 行前 / n 行後の値を参照
sum / average / min / max / count	範囲内の集計

loop（再帰）は非対応

PRQL の loop（SQL の再帰 CTE に相当）構文は、現在の分析エンジンでは利用できません（実行するとサーバエラーになることを実機で確認しました）。 連番の生成は gen_series、前後の行の参照は上記 window の lag / lead で代替してください。

---

組み込み関数カタログ

分析データストアの関数を呼び出すための 組み込み関数 が用意されています。 利用者はそのまま PRQL から呼び出せます。

マップ・JSON 操作

関数	説明	使用例
get_map col key	Map 型フィールドから値を取得	derive {val = get_map tags "severity"}
json_extract_string col key	JSON 文字列からキーの値を取得	derive {name = json_extract_string raw_json "name"}
array_expand field	JSON 配列を行に展開	derive {item = array_expand items}
make_json_kv k v	キーと値から JSON を生成	derive {j = make_json_kv "key" value}
make_json_kv2 k1 v1 k2 v2	2 つの KV から JSON を生成	derive {j = make_json_kv2 "ip" src_ip "port" port}
to_json_array col	配列を JSON 文字列に変換	derive {json = to_json_array arr}

配列操作

関数	説明	使用例
array_join col	配列の要素を行に展開	derive {ip = array_join ip_list}
array_join_newline arr	配列を改行区切りで結合	derive {text = array_join_newline messages}
gen_series n	0 から n-1 の連番（配列）を生成	derive {seq = array_join (gen_series 10)}

Map 生成

関数	説明	使用例
make_map k v	キーと値から Map を生成	derive {m = make_map "action" action}
make_map2 k1 v1 k2 v2	2 つの KV から Map を生成	derive {m = make_map2 "ip" ip "port" port}

文字列・正規表現

関数	説明	使用例
str_length col	文字列長	derive {len = str_length message}
like_pattern col pat	LIKE 検索（% ワイルドカード）	filter (like_pattern path "%/admin/%")
not_like_pattern col pat	NOT LIKE	filter (not_like_pattern path "%/health%")
regex_replace_one col pat repl	正規表現置換（最初の 1 件）	derive {x = regex_replace_one msg "[0-9]+" "#"}
regex_replace_all col pat repl	正規表現置換（全件）	derive {x = regex_replace_all msg "[0-9]" "#"}

条件・IN リスト

関数	説明	使用例
cond_if cond a b	条件分岐（if(cond, a, b)）	derive {lv = cond_if (sev == "high") 3 1}
in2 col a b	いずれかに一致（2 値）	filter (in2 status "open" "resolved")
in3 col a b c	いずれかに一致（3 値）	filter (in3 sev "low" "medium" "high")
in4 col a b c d	いずれかに一致（4 値）	filter (in4 code "A" "B" "C" "D")

型変換

関数	説明	使用例
to_string col	文字列に変換	derive {s = to_string port}
to_int32 col	Int32 に変換	derive {n = to_int32 port}
to_float64 col	Float64 に変換	derive {f = to_float64 score}
to_int64_or_null col	Int64 に変換（失敗時 NULL）	derive {p = to_int64_or_null port}

型は Pipeline 側で指定するのが基本

これらの変換関数は、型を指定していない（文字列の）フィールドをクエリ内でその場変換するためのものです。 Pipeline の Parse Fields で dst_port(Int32) のように型を併記しておけば、dst_port == 22 や sum dst_port を変換なしで実行できます。 詳細は Pipeline（ログ取り込み） を参照してください。

日時操作

関数	説明	使用例
now	現在時刻	derive {current = now}
to_interval_minute n 〜 to_interval_year n	N 分〜N 年のインターバル	filter __time > (now) - (to_interval_day 7)
to_start_of_minute / _hour / _day col	時刻の切り捨て（バケット化）	group {h = to_start_of_hour __time} (...)
to_minute / to_hour / to_day col	分 / 時 / 日（数値）を取得	derive {h = to_hour __time}
to_iso col	ISO 8601（UTC）文字列に変換	derive {ts = to_iso __time}
parse_datetime s	文字列を日時に変換（失敗時 NULL）	derive {d = parse_datetime raw_ts}
to_date_literal s / to_datetime_literal s	文字列リテラルを日付 / 日時に変換	filter __time > (to_datetime_literal "2025-01-01 00:00:00")

to_interval_* は minute / hour / day / week / month / year の 6 種類があります。

ネットワーク分析

関数	説明	使用例
check_is_private_ip ip	プライベート IP か判定	derive {is_private = check_is_private_ip src_ip}
network_graph tbl src dst	ネットワークグラフ JSON を生成	可視化で使用
make_node_colored id color / make_link_proto src dst proto	グラフのノード / リンクを生成	可視化で使用

---

実行例（結果つき）

実データに対する実行結果の例です（テナントのデータ量により値は変わります）。

脆弱性の深刻度別件数

from `vulnerabilities`
group {severity} (
  aggregate {cnt = count this}
)
sort {-cnt}

結果（例）:

severity	cnt
HIGH	663,469
MEDIUM	493,831
CRITICAL	165,021
LOW	31,827
NONE	29

脅威インジケータ（IOC）の脅威レベル別件数

from `indicators`
group {threat_level} (
  aggregate {cnt = count this}
)
sort {-cnt}

結果（例）:

threat_level	cnt
high	557,480
medium	140,416
low	199,574

ユニーク数の集計

from `messages`
aggregate {
  threads = count_distinct thread_id,
  users = count_distinct user
}

結果（例）:

threads	users
1	2

---

実用クエリ例

Top 10 ブロック元 IP

from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {cnt = count this}
)
sort {-cnt}
take 10

直近 1 時間のエラーログ

from `app_logs`
filter __time > (now) - (to_interval_hour 1)
filter level == "ERROR"
sort {-__time}

Tags フィールドから severity を取得して絞り込み

from `alerts`
derive {severity = get_map tags "severity"}
filter severity == "high"
sort {-__time}
take 50

外部（非プライベート）IP からの通信を集計

from `firewall_logs`
derive {is_internal = check_is_private_ip src_ip}
filter is_internal == false
group {src_ip} (
  aggregate {cnt = count this}
)
sort {-cnt}
take 20

JSON フィールドから値を抽出して集計

from `api_logs`
derive {user_id = json_extract_string payload "user_id"}
filter user_id != null
group {user_id} (
  aggregate {request_cnt = count this}
)
sort {-request_cnt}
take 10

送受信ペアごとの通信量

from `firewall_logs`
group {src_ip, dst_ip} (
  aggregate {
    cnt = count this,
    total_bytes = sum bytes
  }
)
sort {-total_bytes}
take 20

---

システムテーブルの参照

ケース・脅威情報・チャットなどの内部データはシステムテーブルとして PRQL で検索できます。 各テーブルのフィールド一覧とクエリ例は システムテーブル を参照してください。

---

セキュリティ制約

PRQL クエリには次のセキュリティ制約が適用されます。

制約	説明
テーブルアクセス制御	自組織の Pipeline、または共有された Pipeline のみアクセス可能
生クエリの埋め込み制限	任意の生クエリ注入を防ぐため、s"..." の使用は now() と toInterval*() のみ許可
禁止関数	ファイル読み取り系関数（*read*）は使用不可
テーブル名の自動プレフィックス	テーブル名には組織を識別するプレフィックスが自動付与され、他組織のデータにはアクセスできない

---

レスポンス形式

Search API（POST /api/v2/user/<user_id>/search/）のレスポンスは、列指向のバイナリ形式（Apache Arrow）で返されます。 Web UI では自動的にデシリアライズされます。 API から直接利用する場合は、Web と同じセッション（Cookie 認証）で次のように処理します。

import pyarrow as pa
import requests

# セッション（Cookie）を取得済みの requests.Session を想定
response = session.post(
    "https://<your-domain>/api/v2/user/<user_id>/search/",
    json={"expr": "from `firewall_logs` take 10"},
)

table = pa.ipc.open_stream(response.content).read_all()
df = table.to_pandas()
print(df)

クエリにエラーがある場合は、{"message": "..."} 形式の JSON がステータスコード 400 で返されます。

---

Sigma ルール変換

Sigma 検知ルール（YAML 形式）を CSIRT-Pro のクエリ形式に変換できます。

POST /api/sigma

Sigma ルールを文字列として送信すると、対応する PRQL クエリが返されます。