はじめに

SplunkのSPLでループ処理を行う方法を調査している中で一つ成功したものがあるのでメモです。
便利だけど初見で理解が難しい以下のコマンドの解説もします。

• streamstats
• eventstats
• foreach

エラトステネスの篩とは

指定された整数以下の素数を発見する、古代ギリシアのエラトステネスさんが編み出した方法です。ここでは詳細は立ち入らずSPLの解説でロジックも見ていきたいと思います。
※Splunkでなんとか実現するために若干ロジックを変えてます。お許しくださいませ。

エラトステネスの篩

SPL

いきなりですが完成したSPLがこちらです。

| makeresults count=120
| streamstats count as num
| eventstats count as max_num
| eval sqrt = sqrt(max_num)
| eval is_prime = if(num=1, "false", "true")
| eval z_loop_{num} = ""
| foreach z_loop_* [| eval <<FIELD>> = if(<<MATCHSTR>> > 1 AND <<MATCHSTR>> < sqrt AND num != <<MATCHSTR>> AND num % <<MATCHSTR>> = 0, 1, <<FIELD>>)]
| foreach z_loop_* [| eval is_prime = if(<<FIELD>> = 1, "false", is_prime)]
| where is_prime = "true"
| table num
| sort num

1行目のcount=で指定した整数（サンプルでは120）以下の素数が表示されます。

それでは中身を見ていきましょう。

篩にかけるための数値を生成

| makeresults count=120

makeresultsにより_timeのみの行が出力されます。デフォルトは1行で、オプションのcount=120を与えることで120行出力しています。

| streamstats count as num

streamstatsは行を上から順に、1行目から対象行までについてstatsを行い、結果を新たな列に出力するためのコマンドです。
つまり、上記のSPLでは毎行countを行っており、例えば1行目では1行目だけを対象なのでcount=1、3行目では3行目までを対象とするのでcount=3をnumに出力します。その結果、連番が得られます。
statsなのでavg、sumも使えます。例えばsumの場合は同様な考えで累計値を出せます。
デフォルトでは対象を1行目からにしていますがwindow=10のようにオプションを与えてあげると現在行含めて10行分が対象になります。これで移動平均を出したりできます。

さて、これで篩にかけるための数値を用意できました。

ループ上限を取得

エラトステネスの篩では、ループを1から指定された整数の平方根まで回します。上限値である平方根を生成してみましょう。

| eventstats count as max_num
| eval sqrt = sqrt(max_num)

eventstatsはstreamstatsと同じく毎行stats処理していきますが、異なるのはstats対象が全行だということです。
そのため毎行、eventstats countにより全行のcountを取得しmax_numに出力しています。
その後、evalのsqrt関数により平方根を計算しています。
SPLでは値を一時保存する変数はありませんので代わりにこのように列に値を保存する必要があります。

準備ができましたので、次に篩にかけていきましょう。

ステップ 1

ステップ1：120要素の配列の1番目にfalseを、2番目以降に全てtrueを入れる。

これは単純ですね。特に説明不要かと思います。

| eval is_prime = if(num=1, "false", "true")

ステップ 2～ステップ3

ステップ 2：配列の先頭から順に走査し、trueの要素を見つけたらその添字pを素数リストに追加し、配列のp^2以上のpの倍数番目をfalseにする。
ステップ 3：上記の篩い落とし操作を、走査している要素の添字がxの平方根に達するまで行う。

ここから楽しくなります。ステップ2と3を一緒にやっちゃいます。

| eval z_loop_{num} = ""

ステップ3に関する1から平方根までのループ処理を行うための準備です。
evalで{}を使うことで、対象行の値を列名にばらして列を追加できます。
numを指定しているため、その値である1～120についてz_loop_を接頭語に列を生成します。
※z_をつけているのは見やすさのため生成した列を右側に表示したかったからです

それではループさせましょう。

| foreach z_loop_* [| eval <<FIELD>> = if(<<MATCHSTR>> > 1 AND <<MATCHSTR>> < sqrt AND num != <<MATCHSTR>> AND num % <<MATCHSTR>> = 0, 1, <<FIELD>>)]

foreachがSPLでほぼ唯一使えるループ処理コマンドです。
foreach z_loop_*により処理対象の列名をワイルドカードで指定できます。これにより、各行について、指定された列名のそれぞれに対して処理を行っていきます。
※つまり、120行×120列の14,400回の処理を行います。

[]内に処理内容を記載します。ここで処理対象の列名を以下で取得できます。

・<<FIELD>>は一致した列名の全文字を示します。z_loop_1などです。
・<<MATCHSTR>>は*で一致した部分の文字列を示します。z_loop_の後の1などです。
そうです、<<MATCHSTR>>で数値が得られるので、これを処理に使えます。

[]内の処理で、1～平方根までの値について、対象の数値で割り切れる場合に処理中の列に1を立てています。（厳密に言うと「平方根まで」は<<MATCHSTR>> < sqrtの条件で判定していますので、平方根以上の数値も処理はしています）
結果は以下のようになります。

z_loop_2の列ではnumが4、6、8（つまり2^nの場合）にフラグを立てています。同様にz_loop_5ではnumが10、15、20にフラグが立っています。
z_loop_*のどこかでフラグが立っていれば、それは対象列の数値で割り切れることになりますので素数ではないことを意味します。

再度foreachを使い、各行において、z_loop_*のいずれかでフラグが立っている場合にis_primeにfalseを出力しています。

| foreach z_loop_* [| eval is_prime = if(<<FIELD>> = 1, "false", is_prime)]

もう完成したも同然ですね！

ステップ 4

ステップ 4：最後までtrueだった要素の添字を素数リストに追加して処理終了。

| where is_prime = "true"
| table num
| sort num

is_primeがtrueだけの行だけを残し、列を絞り、ソートしておしまいです。

応用

以下のように行をappendして列にばらしてあげれば今回のようなループ処理をいつでも使えます。

| append [| makeresults count=10 | streamstats count as tmp]
| eval z_loop_{tmp} = ""
| where isnull(tmp)
| fields - z_loop_, tmp

まとめ

streamstats、eventstats、foreachを使って1～nまでのループ処理を強引に実現してみました。
何かに使えるかも？

（おまけ）処理速度

・・・まあSplunkさんには無理させてるからしゃーない