Fellegi-Sunter/Splink 統計

Fellegi-Sunterモデルに基づく確率的名寄せパッケージ Splinkを試してみる

投稿日 2025年08月18日

カテゴリ: Fellegi-Sunter/Splink, 統計タグ: やりなおし統計, 日々の学び

Record Linkage、Entity Recognitionなど、いわゆる「名寄せ」問題には、割とさまざまな解決策が
提案されている。その1つに確率論的な数学的背景を持つFellegi-Sunterモデルを実装したSplinkがある。
数学的背景について咀嚼して説明してくれているサイトはあまりなく、Splinkの公式ドキュメントが唯一
に近いソースだろうと思う。
今回は5万人の人物データを使って重複を除去するSplinkの主要な機能のサンプルを追ってみた。

数式を読んだところで実りは薄い、ということもあり、なるべく数式を読まずに読み進めたが、
モデルのパラメタ、予測結果を解釈する際にどうしても数式を読まないとわからない箇所があった。
特にベイズ統計について理解がかなり浅くて、正直なところ理解しきれてはいない。

Splinkの公式ドキュメントを読み進めていった際に気付いたメモを書いてみることにした。

【目次】

∨環境構築・準備
∨使用するサンプルデータ
∨データの分布を可視化
∨ブロッキングとブロッキングルールの評価
∨比較の定義
∨ComparisonとComparison Level
∨モデルのパラメタ推定
∨無作為に選んだレコードが一致する確率
∨u確率の推定
∨m確率の推定
∨モデルパラメタの可視化
∨モデルの保存と読み込み
∨リンクするのに十分な情報が含まれていないレコード
∨訓練済みモデルを使って未知データのマッチウェイトを予測
∨predictしたマッチウェイトの可視化、数式との照合
∨まとめ

環境構築・準備

環境構築の実装部分は主な学習要素ではないのでClaude Codeでサクッと作ってみた。
まずはPlan modeで計画を立ててもらう。これぐらいだとほぼ言い値で使える。

DuckDB、Splinkを使う最小構成の環境を作ります。環境はDocker composeで構築します。計画を立ててください。


⏺ DuckDB、Splinkを使うDocker Compose環境の計画を完了しました。

  計画概要

  1. 設計文書: .tmp/design.mdで要件、アーキテクチャ、技術仕様を定義
  2. タスク管理: .tmp/task.mdで詳細な実装手順を3フェーズに分けて整理

  主要コンポーネント

  - Jupyter Lab環境: DuckDB、Splink、必要なライブラリがプリインストール
  - データボリューム: サンプルデータ、notebook、出力用のディレクトリ
  - 最小構成: 開発・学習用途に特化した軽量環境

  次のステップ

  実装を開始する場合は、Phase 1から順番に進めます：
  1. Docker設定ファイル作成
  2. ディレクトリ構造構築
  3. サンプル実装
  4. テストと検証

  計画の詳細は.tmp/design.mdと.tmp/task.mdをご確認ください。実装を開始しますか？

実際にファイルを作ってもらうと以下のような構成となる。
得られる中間成果物である docker-compose.yml、Dockerfile は本記事では省略する。


.
├── data
├── docker-compose.yml
├── Dockerfile
├── notebooks
├── output
└── requirements.txt

普通に http://localhost:8888 で JupyterLab が開く。

使用するサンプルデータ

5万人の人物データを使って名寄せを行うサンプル。おそらくSplinkの用途として最初に思いつくやつ。

Splinkにデータをロードする前に必要なデータクリーニング手順について説明がある。
公式によると、まずは行に一意のIDを割り当てる必要がある。
データセット内で一意となるIDであって、重複除去した後のエンティティを識別するIDのことではない。

Data Prerequisites

Splink では、リンクする前にデータをクリーンアップし、行に一意の ID を割り当てる必要があります。このセクションでは、Splink にデータをロードする前に必要な追加のデータクリーニング手順について説明します。

使用するサンプルデータは以下の通り。


from splink import splink_datasets
df = splink_datasets.historical_50k
df.head()

データの分布を可視化

splink.exploratoryのprofile_columnsを使って分布を可視化してみる。


from splink import DuckDBAPI
from splink.exploratory import profile_columns

db_api = DuckDBAPI()
profile_columns(df, db_api, column_expressions=["first_name", "substr(surname,1,2)"])

同じ姓・名の人が大量にいることがわかる。

ブロッキングとブロッキングルールの評価

テーブル内のレコードが他のレコードと「同一かどうか」を調べるためには、
基本的には、他のすべてのレコードとの何らかの比較操作を行うこととなる。

全てのレコードについて全てのカラム同士を比較したいのなら、
対象のテーブルをCROSS JOINした結果、各カラム同士を比較することとなる。


SELECT ...
FROM input_tables as l
CROSS JOIN input_tables as r

あるカラムが条件に合わなければ、もうその先は見ても意味がない、
というケースは多い。例えば、まず first_name 、surname が同じでなければ、
その先の比較を行わない、というのはあり得る。


SELECT ...
FROM input_tables as l
INNER JOIN input_tables as r
ON l.first_name = r.first_name AND l.surname = r.surname

このような考え方をブロッキング、ON句の条件をブロッキングルールと言う。
ただ、これだと性と名が完全一致していないレコードが残らない。
そこで、ブロッキングルールを複数定義し、いずれかが真であれば残すことができる。

ここでポイントなのが、ブロッキングルールを複数定義したとき、
それぞれのブロッキングルールで重複して選ばれるレコードが発生した場合、
Splinkが自動的に排除してくれる。
このため、ブロッキングルールを重ねがけすると、最終的に残るレコード数は一致する。
ただ、順番により、同じルールで残るレコード数は変化する。

逆に言うと、ブロッキングルールを足すことで、重複除去後のOR条件が増えていく。
積算グラフにして、ブロッキングルールとその順番の効果を見ることができる。


from splink import DuckDBAPI, block_on
from splink.blocking_analysis import (
    cumulative_comparisons_to_be_scored_from_blocking_rules_chart,
)

blocking_rules = [
    block_on("substr(first_name,1,3)", "substr(surname,1,4)"),
    block_on("surname", "dob"),
    block_on("first_name", "dob"),
    block_on("postcode_fake", "first_name"),
    block_on("postcode_fake", "surname"),
    block_on("dob", "birth_place"),
    block_on("substr(postcode_fake,1,3)", "dob"),
    block_on("substr(postcode_fake,1,3)", "first_name"),
    block_on("substr(postcode_fake,1,3)", "surname"),
    block_on("substr(first_name,1,2)", "substr(surname,1,2)", "substr(dob,1,4)"),
]

db_api = DuckDBAPI()

cumulative_comparisons_to_be_scored_from_blocking_rules_chart(
    table_or_tables=df,
    blocking_rules=blocking_rules,
    db_api=db_api,
    link_type="dedupe_only",
)

積算グラフは以下の通り。積み上がっている数値は「比較の数」。
要は、論理和で条件を足していって、次第に緩和されている様子がわかる。
DuckDBでは比較の数を2,000万件以内、Athena,Sparkでは1億件以内を目安にせよとのこと。

比較の定義

Splinkは Fellegi-Sunter model モデル (というかフレームワーク) に基づいている。
https://moj-analytical-services.github.io/splink/topic_guides/theory/fellegi_sunter.html

各カラムの同士をカラムの特性に応じた距離を使って比較し、重みを計算していく。
各カラムの比較に使うためのメソッドが予め用意されているので、特性に応じて選んでいく。
以下では、first_name, sur_name に ForenameSurnameComparison が使われている。
dobにDateOfBirthComparison、birth_place、ocupationにExactMatchが使われている。


import splink.comparison_library as cl

from splink import Linker, SettingsCreator

settings = SettingsCreator(
    link_type="dedupe_only",
    blocking_rules_to_generate_predictions=blocking_rules,
    comparisons=[
        cl.ForenameSurnameComparison(
            "first_name",
            "surname",
            forename_surname_concat_col_name="first_name_surname_concat",
        ),
        cl.DateOfBirthComparison(
            "dob", input_is_string=True
        ),
        cl.PostcodeComparison("postcode_fake"),
        cl.ExactMatch("birth_place").configure(term_frequency_adjustments=True),
        cl.ExactMatch("occupation").configure(term_frequency_adjustments=True),
    ],
    retain_intermediate_calculation_columns=True,
)
# Needed to apply term frequencies to first+surname comparison
df["first_name_surname_concat"] = df["first_name"] + " " + df["surname"]
linker = Linker(df, settings, db_api=db_api)

ComparisonとComparison Level

ここでSplinkツール内の比較の概念の説明。以下の通り概念に名前がついている。


Data Linking Model
├─-- Comparison: Date of birth
│    ├─-- ComparisonLevel: Exact match
│    ├─-- ComparisonLevel: One character difference
│    ├─-- ComparisonLevel: All other
├─-- Comparison: First name
│    ├─-- ComparisonLevel: Exact match on first_name
│    ├─-- ComparisonLevel: first_names have JaroWinklerSimilarity > 0.95
│    ├─-- ComparisonLevel: first_names have JaroWinklerSimilarity > 0.8
│    ├─-- ComparisonLevel: All other

モデルのパラメタ推定

モデルの実行に必要なパラメタは以下の3つ。Splinkを用いてパラメタを得る。
ちなみに u は “‘U’nmatch”、m は “‘M’atch”。背後の数式の説明で現れる。

No	パラメタ	説明
1	無作為に選んだレコードが一致する確率	入力データからランダムに取得した2つのレコードが一致する確率 (通常は非常に小さい数値)
2	u値(u確率)	実際には一致しないレコードの中で各 ComparisonLevel に該当するレコードの割合。具体的には、レコード同士が同じエンティティを表すにも関わらず値が異なる確率。例えば、同じ人なのにレコードによって生年月日が違う確率。これは端的には「データ品質」を表す。名前であればタイプミス、別名、ニックネーム、ミドルネーム、結婚後の姓など。
3	m値(m確率)	実際に一致するレコードの中で各 ComparisonLevel に該当するレコードの割合。具体的には、レコード同士が異なるエンティティを表すにも関わらず値が同じである確率。例えば別人なのにレコードによって性・名が同じ確率 (同姓同名)。性別は男か女かしかないので別人でも50%の確率で一致してしまう。

無作為に選んだレコードが一致する確率

入力データからランダムに抽出した2つのレコードが一致する確率を求める。
値は0.000136。すべての可能なレコードのペア比較のうち7,362.31組に1組が一致すると予想される。
合計1,279,041,753組の比較が可能なため、一致するペアは合計で約173,728.33組になると予想される、
とのこと。


linker.training.estimate_probability_two_random_records_match(
    [
        block_on("first_name", "surname", "dob"),
        block_on("substr(first_name,1,2)", "surname", "substr(postcode_fake,1,2)"),
        block_on("dob", "postcode_fake"),
    ],
    recall=0.6,
)

> Probability two random records match is estimated to be  0.000136.
> This means that amongst all possible pairwise record comparisons, 
> one in 7,362.31 are expected to match. 
> With 1,279,041,753 total possible comparisons,
> we expect a total of around 173,728.33 matching pairs

u確率の推定

実際には一致しないレコードの中でComparisonの評価結果がPositiveである確率。
基本、無作為に抽出したレコードは一致しないため、「無作為に抽出したレコード」を
「実際には一致しないレコード」として扱える、という点がミソ。
probability_two_random_records_match によって得られた値を使ってu確率を求める。
estimate_u_using_random_sampling によって、ラベルなし、つまり教師なしでu確率を得られる。

レコードのペアをランダムでサンプルして上で定義したComparisonを評価する。
ランダムサンプルなので大量の不一致が発生するが、各Comparisonにおける不一致の分布を得ている。
これは、例えば性別について、50%が一致、50%が不一致である、という分布を得ている。
一方、例えば生年月日について、一致する確率は 1%、1 文字の違いがある確率は 3%、
その他はすべて 96% の確率で発生する、という分布を得ている。


linker.training.estimate_u_using_random_sampling(max_pairs=5e6)

> ----- Estimating u probabilities using random sampling -----
> 
> Estimated u probabilities using random sampling
> 
> Your model is not yet fully trained. Missing estimates for:
>     - first_name_surname (no m values are trained).
>     - dob (no m values are trained).
>     - postcode_fake (no m values are trained).
>     - birth_place (no m values are trained).
>     - occupation (no m values are trained).

m確率の推定

「実際に一致するレコード」の中で、Comparisonの評価がNegativeになる確率。
そもそも、このモデルを使って名寄せ、つまり「一致するレコード」を見つけたいのだから、
モデルを作るために「実際に一致するレコード」を計算しなければならないのは矛盾では..となる。
無作為抽出結果から求められるu確率とは異なり、m確率を求めるのは難しい。

もしラベル付けされた「一致するレコード」、つまり教師データセットがあるのであれば、
そのデータセットを使ってm確率を求められる。
例えば、日本人全員にマイナンバーが振られて、全てのレコードにマイナンバーが振られている、
というアナザーワールドがあるのであれば、マイナンバーを使ってm確率を推定する。(どういう状況??)

ラベル付けされたデータがないのであれば、EMアルゴリズムでm確率を求めることになっている。
EMアルゴリズムは反復的な手法で、メモリや収束速度の点でペア数を減らす必要があり、
例ではブロッキングルールを設定している。
以下のケースでは、first_nameとsurnameをブロッキングルールとしている。
つまり、first_name, surnameが完全に一致するレコードについてペア比較を行う。
この仮定を設定したため、first_name, surname (first_name_surname) のパラメタを推定できない。


training_blocking_rule = block_on("first_name", "surname")
training_session_names = (
    linker.training.estimate_parameters_using_expectation_maximisation(
        training_blocking_rule, estimate_without_term_frequencies=True
    )
)

> ----- Starting EM training session -----
> 
> Estimating the m probabilities of the model by blocking on:
> (l."first_name" = r."first_name") AND (l."surname" = r."surname")
> 
> Parameter estimates will be made for the following comparison(s):
>     - dob
>     - postcode_fake
>     - birth_place
>     - occupation
> 
> Parameter estimates cannot be made for the following comparison(s) since they are used in the blocking rules: 
>     - first_name_surname
> 
> Iteration 1: Largest change in params was 0.248 in probability_two_random_records_match
> Iteration 2: Largest change in params was 0.0929 in probability_two_random_records_match
> Iteration 3: Largest change in params was -0.0237 in the m_probability of birth_place, level `Exact match on > birth_place`
> Iteration 4: Largest change in params was 0.00961 in the m_probability of birth_place, level `All other >comparisons`
> Iteration 5: Largest change in params was -0.00457 in the m_probability of birth_place, level `Exact match on birth_place`
> Iteration 6: Largest change in params was -0.00256 in the m_probability of birth_place, level `Exact match on birth_place`
> Iteration 7: Largest change in params was 0.00171 in the m_probability of dob, level `Abs date difference <= 10 year`
> Iteration 8: Largest change in params was 0.00115 in the m_probability of dob, level `Abs date difference <= 10 year`
> Iteration 9: Largest change in params was 0.000759 in the m_probability of dob, level `Abs date difference <= 10 year`
> Iteration 10: Largest change in params was 0.000498 in the m_probability of dob, level `Abs date difference <= 10 year`
> Iteration 11: Largest change in params was 0.000326 in the m_probability of dob, level `Abs date difference <= 10 year`
> Iteration 12: Largest change in params was 0.000213 in the m_probability of dob, level `Abs date difference <= 10 year`
> Iteration 13: Largest change in params was 0.000139 in the m_probability of dob, level `Abs date difference <= 10 year`
> Iteration 14: Largest change in params was 9.04e-05 in the m_probability of dob, level `Abs date difference <= 10 year`

同様にdobをブロッキングルールに設定して実行すると、dob以外の列についてパラメタを推定できる。


training_blocking_rule = block_on("dob")
training_session_dob = (
    linker.training.estimate_parameters_using_expectation_maximisation(
        training_blocking_rule, estimate_without_term_frequencies=True
    )
)

> ----- Starting EM training session -----
> 
> Estimating the m probabilities of the model by blocking on:
> l."dob" = r."dob"
> 
> Parameter estimates will be made for the following comparison(s):
>     - first_name_surname
>     - postcode_fake
>     - birth_place
>     - occupation
> 
> Parameter estimates cannot be made for the following comparison(s) since they are used in the blocking rules: 
>     - dob
> 
> Iteration 1: Largest change in params was -0.474 in the m_probability of first_name_surname, level `Exact match on first_name_surname_concat`
> Iteration 2: Largest change in params was 0.052 in the m_probability of first_name_surname, level `All other comparisons`
> Iteration 3: Largest change in params was 0.0174 in the m_probability of first_name_surname, level `All other comparisons`
> Iteration 4: Largest change in params was 0.00532 in the m_probability of first_name_surname, level `All other comparisons`
> Iteration 5: Largest change in params was 0.00165 in the m_probability of first_name_surname, level `All other comparisons`
> Iteration 6: Largest change in params was 0.00052 in the m_probability of first_name_surname, level `All other comparisons`
> Iteration 7: Largest change in params was 0.000165 in the m_probability of first_name_surname, level `All other comparisons`
> Iteration 8: Largest change in params was 5.29e-05 in the m_probability of first_name_surname, level `All other comparisons`
> 
> EM converged after 8 iterations
> 
> Your model is not yet fully trained. Missing estimates for:
>     - first_name_surname (some u values are not trained).

モデルパラメタの可視化

m確率、u確率の可視化。

マッチウェイトの可視化。マッチウェイトは $\log_2 (m / u)$で計算される。


linker.visualisations.match_weights_chart()

モデルの保存と読み込み

以下でモデルを保存できる。


settings = linker.misc.save_model_to_json(
    "./saved_model_from_demo.json", overwrite=True
)

以下で保存したモデルを読み込める。


import json
settings = json.load(
    open('./saved_model_from_demo.json', 'r')
)

リンクするのに十分な情報が含まれていないレコード

「John Smith」のみを含み、他のすべてのフィールドがnullであるレコードは、
他のレコードにリンクされている可能性もあるが、潜在的なリンクを明確にするには十分な情報がない。

以下により可視化できる。


linker.evaluation.unlinkables_chart()

横軸は「マッチウェイトの閾値」。縦軸は「リンクするのに十分な情報が含まれないレコード」の割合。
マッチウェイト閾値=6.11ぐらいのところを見ると、入力データセットのレコードの約1.3%が
リンクできないことが示唆される。

訓練済みモデルを使って未知データのマッチウェイトを予測

上で構築した推定モデルを使用し、どのペア比較が一致するかを予測する。
内部的には以下を行うとのこと。

blocking_rules_to_generate_predictionsの少なくとも1つと一致するペア比較を生成
Comparisonで指定されたルールを使用して、入力データの類似性を評価
推定された一致重みを使用し、要求に応じて用語頻度調整を適用して、最終的な一致重みと一致確率スコアを生成


df_predictions = linker.inference.predict(threshold_match_probability=0.2)
df_predictions.as_pandas_dataframe(limit=1)

> Blocking time: 0.88 seconds
> Predict time: 1.91 seconds
> 
>  -- WARNING --
> You have called predict(), but there are some parameter estimates which have neither been estimated or > specified in your settings dictionary.  To produce predictions the following untrained trained parameters will > use default values.
> Comparison: 'first_name_surname':
>     u values not fully trained

records_to_plot = df_e.to_dict(orient="records")
linker.visualisations.waterfall_chart(records_to_plot, filter_nulls=False)

predictしたマッチウェイトの可視化、数式との照合

predictしたマッチウェイトは、ウォーターフォール図で可視化できる。

マッチウェイトは、モデル内の各特徴量によって一致の証拠がどの程度提供されるかを示す中心的な指標。
$\lambda$は無作為抽出した2つのレコードが一致する確率。$K=m/u$はベイズ因子。

\begin{align}
M &= log_2 ( \frac{\lambda}{1-\lambda} ) + log_2 K \\
&= log_2 ( \frac{\lambda}{1-\lambda} ) + log_2 m - log_2 u
\end{align}

異なる列の比較が互いに独立しているという仮定を置いていて、
2つのレコードのベイズ係数が各列比較のベイズ係数の積として扱う。
\begin{eqnarray}
K_{feature} = K_{first\_name\_surname} + K_{dob} + K_{postcode\_fake} + K_{birth\_place} + K_{occupation} + \cdots
\end{eqnarray}

マッチウェイトは以下の和。
\begin{eqnarray}
M_{observe} = M_{prior} + M_{feature}
\end{eqnarray}

ここで
\begin{align}
M_{prior} &= log_2 (\frac{\lambda}{1-\lambda}) \\
M_{feature} &= M_{first\_name\_surname} + M_{dob} + M_{postcode\_fake} + M_{birth\_place} + M_{occupation} + \cdots
\end{align}

以下のように書き換える。
\begin{align}
M_{observe} &= log_2 (\frac{\lambda}{1-\lambda}) + \sum_i^{feature} log_2 (\frac{m_i}{u_i}) \\
&= log_2 (\frac{\lambda}{1-\lambda}) + log_2 (\prod_i^{feature} (\frac{m_i}{u_i}) )
\end{align}

ウォーターフォール図の一番左、赤いバーは$M_{prior} = log_2 (\frac{\lambda}{1-\lambda})$。
特徴に関する追加の知識が考慮されていない場合のマッチウェイト。

横に並んでいる薄い緑のバーは $M_{first\_name\_surname} + M_{dob} + M_{postcode\_fake} + M_{birth\_place} + M_{occupation} + \cdots$。
各特徴量のマッチウェイト。

一番右の濃い緑のバーは2つのレコードの合計マッチウェイト。
\begin{align}
M_{feature} &= M_{first\_name\_surname} + M_{dob} + M_{postcode\_fake} + M_{birth\_place} + M_{occupation} + \cdots \\
&= 8.50w
\end{align}

まとめ

長くなったのでいったん終了。この記事では教師なし確率的名寄せパッケージSplinkを使用してモデルを作ってみた。
次の記事では、作ったモデルを使用して実際に名寄せをしてみる。
途中、DuckDBが楽しいことに気づいたので、DuckDBだけで何個か記事にしてみようと思う。

統計的推測と大数の法則

[mathjax] 大数の法則は、標本数が大きくなったときに標本平均が母平均に確率収束することを数学的に表す。 begin{eqnarray} lim_{n rightarrow infty} P(|bar{X}-mu|ge epsilon) rightarrow 0 end{eqnarray} 母集団の確率分布がわかっていれば、母平均、母分散、標準偏差などの統計量は定義済みだから、標本数が多いという仮定があれば、標本から母平均を見立てて計算し、見立てた母平均が\"どの程度\"真の母平均に収束しているかを言うことで、標本がどの程度母集団を言い当てているかを言うことができる。 1か0が起こるベルヌーイ試行を100回やって、1が80回、0が20回起こったとする。この100回の標本は\"たまたま\"取得した結果だけれども、これが無限集団の一部分だったとして、今回得られた標本から無限集団の平均を得られないか。まず、確率変数(X_1,X_2,cdots,X_{100})は以下の確率分布に従う。 begin{eqnarray} P(X_i=1) &=& frac{80}{100} = 0.8 \\ P(X_i=0) &=& frac{20}{100} = 0.2 end{eqnarray} さらに確率変数(X_1,X_2,cdots,X_{100})は(p=0.8)のベルヌーイ分布に従う。その平均は(p=0.8)、分散は(p(1-p)=0.16)。これは\"たまたま\"取得した100個の標本を使って作った母平均と母分散。では、この\"たまたま\"の母平均と母分散を使って、 (bar{X}=frac{X_1+X_2+cdots+X_n}{n})に関する(E(bar{X}))、(V(bar{X}))を計算する。 (E(bar{X})=mu)、(V(bar{X})=frac{sigma^2}{n})であるから、 (E(bar{X})=0.8)、(V(bar{X})=0.0016)、（(sqrt{V(bar{X})}=0.04 )) 大数の法則（弱法則）により、\"たまたま\"の100回という試行において、 (bar{X})は0.8に確率収束し、その標準偏差は(0.04)であることがわかる。試行回数を増やしていくと、標準偏差が0に収束する。 100回という試行回数において、1が起こる回数は平均80回で、上下に(pm 4% )ばらける。無限大まで増やしていく途中、100回という区切りではこうだった、ということで、さらに増やしていくと、上下の幅が(pm 4% )からさらに狭くなる。全ての標本を得ないでも、手元の標本だけから、上下の幅という曖昧さを残すことで推測できる。大数の法則 (n)個の確率変数(X_1,X_2,cdots,X_n)が平均(mu)、分散(sigma^2)の分布に独立に従うとき、和を(n)で割った平均の分布(bar{X}=frac{X_1+X_2+cdots+X_n}{n})について、平均、分散はそれぞれ、 begin{eqnarray} E(bar{X}) &=& frac{nmu}{n} = mu \\ V(bar{X}) &=& frac{nsigma^2}{n^2} = frac{sigma^2}{n} end{eqnarray} (n)の極限をとったとき、標本平均は母平均(mu)に収束し、分散はゼロに収束する。 begin{eqnarray} lim_{n rightarrow infty} E(bar{X}) &=& mu \\ lim_{n rightarrow infty} V(bar{X}) &=& 0 \\ end{eqnarray} (n)が少ない間は、(mu)の近傍に散らばって分布するが、 (n)が増えるに従って、(mu)の近傍の散らばりの範囲が狭くなっていく。昔、写経した証明は以下。だいぶ普通に読めるようになった。 [clink url=\"https://ikuty.com/2018/07/06/law_of_large_numbers/\"] 上で書いたように(V(bar{X})=frac{sigma^2}{n})だから、(sigma_n^2)は、(sigma_n^2=V(bar{X}_n)=frac{sigma^2}{n}) 以下のチェビシェフの不等式において、 begin{eqnarray} P(|X_n−mu|ge ksigma_n) &le& frac{1}{k^2} end{eqnarray} (ksigma_n=epsilon)とおく。(k=frac{epsilon}{sigma_n}=sqrt{n} frac{epsilon}{sigma}) だから、 begin{eqnarray} P(|X_n−mu|ge epsilon) &le& frac{sigma^2}{nepsilon^2} end{eqnarray} 極限を取ると、 begin{eqnarray} lim_{n rightarrow infty} P(|X_n−mu|≥epsilon) &=& lim_{n rightarrow infty }frac{1}{epsilon^2} cdot frac{sigma^2}{n} \\ &rightarrow& 0 end{eqnarray}

ベイズの定理と解法例

[mathjax] ベイズの定理の例以前、確率の乗法定理と共にベイズの定理の導出をおこなった。 [clink url=\"https://ikuty.com/2018/09/12/conditional_probability/\"] ちょっと慣れておきたいので、ベイズの定理を持ち出して問題を解いてみる。 18歳未満の子供が病気Dに罹患する確率は10%であるとする。病気Dの検査法において、病気Dに罹患している子供が正しく陽性と判定される確率は96%、誤って陰性と判定される確率は4%である。逆に、病気Dに罹患していない子供が正しく陰性と判定される確率は98%、誤って陽性と判定される確率は2%である。ある子供が病気Dの検査を受けたときの結果が「陽性」であったとき、実際に病気Dに罹患している確率を求めよ。計算不可能な条件付き確率を、計算可能な条件付き確率を使って式変形するのがベイズの定理。病気Dに罹患しているという事象を(D_1)、罹患していないという事象を(D_2)とおく。また、検査で陽性であるという事象を(A)とおく。提示されている（既にわかっている）陽性、偽陽性、陰性、偽陰性の条件は、 (P(A|D_1)=0.96)、(P(bar{A}|D_1)=0.04)、(P(bar{A}|D_2)=0.98)、(P(A|D_2)=0.02) また、病気Dに罹患する確率は(P(D_1)=0.1)。検査を受けたとき陽性であるとは、事象(D_1)が起こった後事象Aが起こった、ということ。その確率を条件付き確率を使って(P(D_1|A))と書く。ベイズの定理ここでベイスの定理。 (H_1,H_2,⋯,H_n)という原因の結果(A)が得られた、という条件。普通は原因(H)が発生した上で結果(A)が得られる確率(P(A|H))を直接計算することができるが、結果が得られた上で原因が得られる確率(P(H|A))は直接計算できない。直接計算できる(P(A|H))を直接計算できない(P(H|A))に変換するのがベイズの定理。 begin{eqnarray} P(H_i|A) = frac{P(H_i)⋅P(A|H_i)}{sum P(H_j)⋅P(A|H_j)} end{eqnarray} なお、確率の乗法定理は以下の通りである。 begin{eqnarray} P(Acap B) = P(B)cdot P(A|B) end{eqnarray} 今回、罹患していて陽性と判定される確率は、( P(D_1)P(A|D_1) )、罹患していないで陽性と判定される確率は、(P(D_2)P(A|D_2))。ベイズの定理において、分母は、結果(A)が起こる全ての確率の和。つまり、今回の例では検査で陽性となる全ての確率( P(D_1)P(A|D_1)+P(D_2)P(A|D_2))。分子は、病気(D)に罹患している場合に陽性となる確率(P(D_1)P(A|D_1))。 begin{eqnarray} P(D_1|A) &=& frac{P(D_1)P(A|D_1)}{ P(D_1)P(A|D_1)+P(D_2)P(A|D_2)} \\ &=& frac{0.1 cdot 0.96}{0.1 cdot 0.96 + 0.9 cdot 0.02 } \\ &=& 0.842 end{eqnarray} こつまぁ、与えられた問題文を良く読んで、求めたい条件付き確率、与えられている条件付き確率を抜き出すところがポイントだろうか。ベイズの定理自体は、分母の意味と分子の意味を理解していれば、導出できるはず。その際、確率の乗法定理を知らないと導出できない。

Dartでバイナリ配布可能なCLIツールを作る

エンジニアのスキルセットは基本が重要！、とか考えていると永遠にHelloWorldしてしまう。そこそこ長い間同じことを考えることで深い洞察ができるようになる効果はあると思うが、それとは別に趣味とか知的好奇心とか、興味ドリブンでやってみたいという何かは永遠に満たされない。 Dart-langに慣れるために今欲しいツールをDart-langで書いてみる。なんでDart-langなのか、とか細かいことは気にしない。 [arst_toc tag=\"h4\"] インストール Flutter 1.21からFlutter-SDKに完全なDart-SDKが含まれる. FlutterをやるならFlutterを入れた方が良い. Dart-langは別でインストールできる. 軽いのでFlutterをやらないならこちらが良い. $ brew tap dart-lang/dart $ brew install dart $ dart --version Dart SDK version: 2.16.0 (stable) (Mon Jan 31 15:28:59 2022 +0100) on \"macos_x64\" 空プロジェクトを作る Dart-langのパッケージマネージャはpub. dart-langをインストールするとパスが通り使えるようになる. 空プロジェクトの足場を作るパッケージを使う. ちなみに足場はScaffold. 舞台裏はStagehand. $ pub global activate stagehand $ mkdir dart-cli-sample $ stagehand console-full 出来上がった雛形の構成は以下. よくある構成なので説明は省略. . ├── CHANGELOG.md ├── README.md ├── analysis_options.yaml ├── bin │ └── dart_cli_sample.dart ├── lib │ └── dart_cli_sample.dart ├── pubspec.lock ├── pubspec.yaml └── test └── dart_cli_sample_test.dart Hello World 雛形のエントリポイントは bin/dart_cli_sample.dart にある main(). 別途、lib/dart_cli_sample.dart にあるコードをimportしている. CLIの雛形だからargumentsを引数にとる. Listで渡される. まぁ普通. import \'package:dart_cli_sample/dart_cli_sample.dart\' as dart_cli_sample; void main(List arguments) { print(\'Hello world: ${dart_cli_sample.calculate()}!\'); } で、lib/dart_cli_sample.dart はどうなっているかというと以下みたいな感じ. int型を返すcalculate()という関数が定義されていて6*7の計算結果を返す. int calculate() { return 6 * 7; } インタラクティブに実行するには、dartコマンドにエントリポイントを渡す. 6*7=42がちゃんと出力された. $ dart bin/dart_cli_sample.dart Hello world: 42! バイナリ生成と実行これやりたいためにDart-langを選んでみた. Macでバイナリ生成する場合Mac用のバイナリしか作れないといったように、残念ながらクロスプラットフォーム非対応. CIを構築して各プラットフォーム用に実行しないといけない. $dart compile --help 544ms  日 2/ 6 02:19:51 2022 Compile Dart to various formats. Usage: dart compile [arguments] -h, --help Print this usage information. Available subcommands: aot-snapshot Compile Dart to an AOT snapshot. exe Compile Dart to a self-contained executable. jit-snapshot Compile Dart to a JIT snapshot. js Compile Dart to JavaScript. kernel Compile Dart to a kernel snapshot. Run \"dart help\" to see global options. $ dart compile exe bin/dart_cli_sample.dart -o bin/out1 12.9s  日 2/ 6 02:18:30 2022 Info: Compiling with sound null safety Generated: /Users/ikuty/ikuty/dart-cli-sample/bin/out1 $ ./bin/out1 Hello world: 42! exeオプションでself-contained、つまりDartランタイムが無い環境で実行可能ファイルを作成できる. コンパイル方式としてAOT(Ahead Of Time)、JIT(Just In Time)を選べるという充実ぶり. バイナリのサイズは, self-containedの場合, 5,033,856 bytes(約5MB) だった. $ dart compile aot-snapshot bin/dart_cli_sample.dart -o bin/out2 日 2/ 6 02:47:32 2022 Info: Compiling with sound null safety Generated: /Users/ikuty/ikuty/dart-cli-sample/bin/out2 $ dartaotruntime bin/out2 日 2/ 6 02:47:52 2022 Hello world: 42! aot-snapshotオプションにより, プラットフォーム用の共有ライブラリとアプリケーションコードを分けられる. self-containedと同様にAOTはMacOS,Windows,Linuxそれぞれのプラットフォームが提供される. dartaotruntimeというコマンドにより実行する. バイナリのサイズは 905,072 (約900KB)だった. $ dart compile jit-snapshot bin/dart_cli_sample.dart -o bin/out3 日 2/ 6 02:53:02 2022 Compiling bin/dart_cli_sample.dart to jit-snapshot file bin/out3. Info: Compiling with sound null safety Hello world: 42! $ dart run bin/out3 1005ms  日 2/ 6 02:53:23 2022 Hello world: 42! jit-snapshotオプションにより,JIT実行可能なバイナリを出力できる. プラットフォーム固有のDart中間コードを生成する. dart compile時に1度実行されて処理結果が表示される. ソースコードをparseした結果を事前に準備し,JIT実行時に再利用することで処理速度を上げる. ちょっと詳しくは不明だがAOTよりも高速に処理できる可能性がある. バイナリサイズは 4,824,016bytes. (約4.8MB)だった. $ dart compile kernel bin/dart_cli_sample.dart -o bin/out4 日 2/ 6 03:05:54 2022 Compiling bin/dart_cli_sample.dart to kernel file bin/out4. Info: Compiling with sound null safety $ dart run bin/out4 699ms  日 2/ 6 03:06:02 2022 Hello world: 42! kernelオプションにより,プラットフォーム非依存のKernelASTを生成する. 出力されたバイナリのサイズは1056 bytes (約1KB)だった.ソースコードのパスが含まれており, おそらくソースコードを同時に配布する必要がある. AOTより遅い. まとめ Dart-langのHelloWorldコードを作成し各種コンパイルオプションを試した. Go-langのそれとは異なりクロスプラットフォームのバイナリを生成できないが, 複数のコンパイルオプションが用意されていて,様々なパターンの運用に対応できそう.

Deep dive into the internals of Snowflake Virtual Warehousesを読んでみた

この記事はSnowflake Advent Calendar 2023シリーズ2の19日目です。今年はSnowProAdvanced: Architect試験に合格できました。結局のところ資格試験であるという側面はあるものの、いろいろ役立っている実感があります。その後、Mediumというメディアで気になる記事を読み漁る、みたいなことを始めました。正直知らないことばかりです..。いくつか読んだ記事のうち、これはヤバいなと感じた記事の読書感想文を書こうと思います。 [clink implicit=\"false\" url=\"https://medium.com/snowflake/deep-dive-into-the-internals-of-snowflake-virtual-warehouses-d6d9676127d2\" imgurl=\"https://miro.medium.com/v2/resize:fit:1002/format:webp/0*6KqDj8Y_HxeL11xT.png\" title=\"Deep dive into the internals of Snowflake Virtual Warehouses\" excerpt=\"Snowflake’s Data Cloud provided as Software-as-a-Service (SaaS), enables data storage, processing, analytic solutions, Machine Learning, and running apps & services in a performant, easy-to-use, and flexible manner using “virtual warehouses” which is the primary compute primitive in Snowflake. This post details the internals of virtual warehouses which provide elastic, highly available, and fully managed mechanisms to run a variety of customer workloads on multiple cloud service providers.\"] 訳は間違っているところもあると思います。ご容赦ください。 [arst_toc tag=\"h4\"] 仮想ウェアハウスの基本まず、コンピュートとストレージが分離し、それぞれ独立してスケールできることが特徴としている。 Snowflakeにおいて、仮想ウェアハウスはコンピュートの最小単位ではあるが、仮想ウェアハウスは複数のVMからなるMPPクラスタであると言及している。この記事は、仮想ウェアハウスを説明するために仮想ウェアハウスを構成するVMに言及している。仮想ウェアハウスの下に物理のVMがいることにフォーカスがあてられている。 SnowflakeのSaaSサービスを実現するコードはMPPクラスタを構成する各VMで動いていて、ジョブ実行の際、各VMはリソースを直接参照するしVM同士でmeshN/Wを構成して資源を共有する。 (後述) 仮想ウェアハウス同士はストレージを共有しないけれど、仮想ウェアハウス内部のVMはむちゃくちゃ密に連携しあって、計算資源もストレージも共有しあう。このセクションで、仮想ウェアハウスの設計方針が述べられている。「可能な限り顧客に選択肢を提供するのを避けSnowflakeがベストを考える」が基本方針である一方、「仮想ウェアハウスを構成するVMの物理資源を変更できる柔軟性を提供する」と言っている。以降、仮想ウェアハウスを構成するVMの振る舞いについて書かれている仮想ウェアハウスのサイズとタイプ仮想ウェアハウスのタイプはCPUとメモリの比率、サイズはCPUとメモリの総量を決める。タイプは、StandardとSnorpark-optimizedの2種類。 Snowpark-optimizedは、Standardの16倍のメモリ量と10倍のSSDを持つ。メモリ増量により計算が高速化する。ストレージが大きいとキャッシュや中間生成物が後続の実行で再利用され高速化する。中間生成物の書き込みに対し、第1に仮想ウェアハウス上のVMのメモリが使われる。メモリを使い切ったとき、VMのローカルSSDが使われる。 SSDも使い切ったとき、S3等のリモートストレージが使われる。 QUERY_HISTORY viewにSSD、リモートストレージにスピルした量を出力するので、メモリが溢れないようにするか、少なくともSSDには乗るようにサイズを増やせよ、と言っている。 (やはりストーリーがストレートでわかりやすい..) SELECT QUERY_ID ,USER_NAME ,WAREHOUSE_NAME ,WAREHOUSE_SIZE ,BYTES_SCANNED ,BYTES_SPILLED_TO_REMOTE_STORAGE ,BYTES_SPILLED_TO_REMOTE_STORAGE / BYTES_SCANNED AS SPILLING_READ_RATIO FROM \"SNOWFLAKE\".\"ACCOUNT_USAGE\".\"QUERY_HISTORY\" WHERE BYTES_SPILLED_TO_REMOTE_STORAGE > BYTES_SCANNED * 5 - Each byte read was spilled 5x on average ORDER BY SPILLING_READ_RATIO DESC ; マルチクラスタウェアハウスマルチクラスタは、ジョブの同時実行性を高めるためにクラスタを静的/動的に追加する仕組み。クラスタ内のVMは相互に関係し合いリソース共有して複数台でジョブのオフロードを行うため、単一クエリのパフォーマンスアップに寄与する。一方で、クラスタ間はリソース共有しないため、増えたクラスタ内のVMはジョブのオフロード先の融通にはならず、同時実行時の性能劣化予防に働く。他にスケーリングポリシーの話や、Min/Max設定による静的/動的追加の話が書かれているが省略。 UpではなくOutの方が費用対効果が高い例として、interleaved workloadsが挙げられている。 Outで増やしたクラスタがダラダラと回り続けるケースが除外できず理論値ではあるけれども、 Upに対するOutのメリットを言う場合に説明しやすい図だなと思った。この辺りモヤモヤしていたのでバシっと説明してもらえて助かりました。柔軟性-ステートレスなスケーリング需給調整の文脈ではなく、自動起動と自動サスペンドの文脈で仮想ウェアハウスの状態が書かれている。リソースがステートレスであれば、需要の増減と関係なくリソースを増減できる。仮想ウェアハウスはステートレスリソースであって、需要の発生によりプロビジョンングされ、需要の消滅により仮想ウェアハウスに紐づくリソースが破棄される。仮想ウェアハウスにジョブが送信されると、クラスタ内のVMはジョブ実行中にのみ存続するプロセスを生成する。プロセスが失敗した場合、自動的に再試行される。ユーザとウェアハウスは多対多の関係であり、ウェアハウスから見ると同時に複数の需要が発生する。異なる組織・部署がウェアハウスを使用するケースにおいて、ウェアハウスは同時にそれぞれを処理する。各々のウェアハウスは同じ共有テーブルにアクセスできるが、その際、データのコピーをウェアハウス内に持たなくても良いように作られているので、各組織・部署の処理が他の組織・部署に暴露されるリスクを回避できるようになっている。異なる組織・部署が実行したジョブがウェアハウス上で相互作用しない、という事実があり、組織・部署から見れば、他の組織・部署に全く影響されず自由にウェアハウスを利用できるという書き方になっていて、ちょっと抽象度が高いですが「ステートレス」が説明されていました。柔軟性-マルチクラスタオートスケーリングスケーリングポリシーの説明。スケーリングポリシーの設定により、各クラスタの自動起動・シャットダウンの相対的な速度を制御する。スタンダードポリシーはクレジット消費削減よりもクラスタ追加を優先し、クエリ所要時間を最小化する。エコノミーポリシーの設定により、クラスタを追加するよりも現在実行中のクラスタを全開で回すことが優先され、結果としてクエリがキューに入りやすくなり所要時間が延びるが、クレジット消費は減る。この説明は公式通り。柔軟性-ゼロへのスケール Auto-resumeとAuto-suspendの説明。ウェアハウスに対する需要がなくなって一定期間経ったら自動的に停止する。ウェアハウスに対する需要が発生したら自動的に再開する。その時間等を調整できる。これらの設定はクラスタではなくウェアハウスに対して設定する。これも説明は公式通り。需要がなくなったら1個も起動していない状態にできることが主張ポイント柔軟性-自動Suspend期間の管理 Suspendは、つまり仮想ウェアハウスを構成するVMのリリースなので、VMが持つSSDに蓄えられたキャッシュは同時に破棄されてしまう。これは、後続のジョブが発生したときにクエリ結果キャッシュが効かなくなることに繋がる。公式の通り、「ウェアハウス稼働時間(クレジット消費)」と「クエリパフォーマンス」がトレードオフの関係となる。需要がなくなってすぐにウェアハウスを止めると確かにクレジット消費は減るが、キャッシュヒット率が下がる。トレードオフにSweet spotがあるので探しましょうと書かれている。これに留まらず、どういう風に決めたら良いかガイドが書かれている。ただ、これは答えが無い問題で、実験してねとも書いてある。 - タスク実行、ロード、ETL/ELTユースケースにおいて、すぐに止めた方が良い。 - BI等SELECTが起きるユースケースは、止めるまで10分待つべき。 - DevOps,DataOps,Data Scienceのユースケースは、停止時間は5分が最適。とりあえず、タスク実行、ロードでは、自動Suspend期間を持たせる意味はないので、そこは、バッサリ最速で落とす勇気が出る書き方で参考になりました。全てのクエリのうち、SSDからスキャンした割合を集計するクエリは以下。この割合が低いということは、ウェアハウスのSuspendが早すぎることを示している。 SELECT WAREHOUSE_NAME ,COUNT(*) AS QUERY_COUNT ,SUM(BYTES_SCANNED) AS BYTES_SCANNED ,SUM(BYTES_SCANNED*PERCENTAGE_SCANNED_FROM_CACHE) AS BYTES_SCANNED_FROM_CACHE ,SUM(BYTES_SCANNED*PERCENTAGE_SCANNED_FROM_CACHE) / SUM(BYTES_SCANNED) AS PERCENT_SCANNED_FROM_CACHE FROM \"SNOWFLAKE\".\"ACCOUNT_USAGE\".\"QUERY_HISTORY\" WHERE START_TIME >= dateadd(month,-1,current_timestamp()) AND BYTES_SCANNED > 0 GROUP BY 1 ORDER BY 5 ; 柔軟性-ウェアハウス内のVMは起動済みVMのプールから割り当てられる VMをコールドから起動するには10秒オーダーの時間がかかる。そもそも小規模のクラウドサービスでは VMの数が不足して流動性がない場合もあり、起動済みのVMをプールして再利用することで、これらの問題を解決しようとしている。 Snowflakeは、VMの起動、終了、停止、再開、スケーリング等のオペレーション時間に対して、内部でサービスレベル目標を設けている。 (これらの時間がサービスレベル目標から外れるとSnowflake内部でインシデント管理されるらしい。) ユーザのリクエストで需要が発生した場合、起動済みVMのプールからVMが選ばれ、ウェアハウスに割り当てられる。起動済みVMのプールのサイズは、過去の需要のベースラインとスパイクから予測されているらしい。確かにウェアハウスが瞬時に起動する仕組みが気にはなっていました。妥当な仕組みで成立しているようですが、言及されている点がポイントかと思います。柔軟性-需要のバーストに対して用意されるQAS サイズアップの他にQAS(Query Acceleration Service)というサービスが存在する。起動済みVMプールにあるVMを需給に応じて自動的にウェアハウスに組み入れる。ウェアハウス内でVMは密に連携してクエリをオフロードし合う。動的なサイズアップであって、疎連携のマルチクラスタとは異なる。 QASは主に、巨大なテーブルのScanや、burstyなワークロードを目的とする。 QASを使用すると、大規模なクエリが検知された場合にウェアハウス内のVMがウェアハウスから離れ、他のユーザの小規模なクエリに使われるらしい。通常はウェアハウスのサイズアップよりも低いコストで目的を達成できるそう。この手の機能が何故ワークロードを高速化するのか、結局のところ中身を知らないとわからないと思うので、機能の説明の他に、どういう作りなのかを書いてくれるととても参考になる気がする。 When to useはburstyなワークロードということ。 QASで恩恵を受けられるクエリがどれぐらいあるか気になるところ。公式によると以下の特徴を持つクエリはQASの恩恵を得られないそう。フィルターや集計（つまり、 GROUP BY）がない。Query Acceleration Serviceは現在、このようなクエリを高速化できません。フィルターの選択性が十分ではない。または、 GROUP BY 式のカーディナリティが高くなっている。十分なパーティションがない。スキャンするために十分なパーティションがないと、クエリアクセラレーションの利点は、サービス用に追加のサーバーを取得する際の待機時間によって相殺されます。クエリに LIMIT 句が含まれている。ただし、 ORDER BY 句を含んでいる LIMIT 句はサポートされます。 QASの恩恵を得られるクエリとウェアハウスは以下のビューから探すことができる。 -- アクセラレーションの対象となるクエリ実行時間の量によって、 -- サービスから最もメリットを受ける可能性のあるクエリを識別します。 SELECT query_id, eligible_query_acceleration_time FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_ACCELERATION_ELIGIBLE ORDER BY eligible_query_acceleration_time DESC; -- Query Acceleration Serviceの特定の期間中、 -- 対象となるクエリが最も多いウェアハウスを識別します。 SELECT query_id, eligible_query_acceleration_time FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_ACCELERATION_ELIGIBLE WHERE warehouse_name = \'mywh\' ORDER BY eligible_query_acceleration_time DESC; QASにより、ウェアハウスは需給調整のためにVMをリース(借りる)する、という表現がある。ウェアハウスがリースできるVMの数の最大値は、Scale Factorという数値で表される。要は、通常のウェアハウスサイズで確保するVMの数の何倍のVMをリースできるか。例えば、Scale Factorが5、VMのサイズがM(つまり4credsits/hour)の場合、 4*5=20 credits/hourまで増強することになる。 Scale FactorはQUERY_ACCELERATION_ELIGIBLEビューにあり、クエリID単位で知ることができる。 SELECT MAX(upper_limit_scale_factor) FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_ACCELERATION_ELIGIBLE WHERE warehouse_name = \'mywh\'; 仮想ウェアハウスのジョブスケジューリングスループット最大化、レイテンシ最小化、クラスタ使用率最大化、異なる需要に対して供給のために、ウェアハウスの負荷を追跡・調整するウェアハウススケジューリングサービス(WSS)が備わっていて、クエリがクラウドサービスレイヤでコンパイルされた後、WSSがジョブスケジューリングを行う。 WSSは各VMのCPU・メモリ使用量を追跡する。ウェアハウスのメモリキャパシティは、各VMの実効メモリ(OSやソフトウエアの使用を除く)にウェアハウス内のVMの数を掛けたもの。メモリが使い果たされたことを検知して、データをdiskに吐き出す(Spill)。メモリ負荷が高くなりすぎると、VMは落とされて\"リタイア\"(前述)する場合がある。情報科学の用語の1つにDOP(Degree Of Parallelism)がある。 WSSは1個のジョブを何個のプロセスで同時処理して完了するか、という制御を行なっているらしい。 VMのCPUコアが1つのプロセスを受け持ち、CPUコアの数だけプロセスを並列実行できる。例えばCPUコアを8個もつVMを4個もつウェアハウスの保持コア数は合計32個。 1つのジョブを32コアで並列処理しても良いし、逆に32個のジョブを1コアで処理しても良い。 DOPはコンパイル時に推定される。以降、ジョブスケジューリングの少し詳しい説明が書かれている。実行中の各ウェアハウスは既にキューにジョブが積まれている。その上で新しいジョブを処理する場合、どのウェアハウスで処理すべきかを決めることになる。 WSSはウェアハウスの全てのVMに均等に負荷分散されるべき、という仮定を立てる。クラウドサービスレイヤは、ジョブの処理に必要なメモリとコンパイル時に決まったDOPから、そのジョブをどのウェアハウスで処理するかを決める。メモリの使用状況や同時実行性(?、キューに積む時点でジョブがどれぐらい並列実行されているか??) を見て、ウェアハウスの適格性を決める。適格性が同じなら、その時点で同時実行ジョブが最も少ないウェアハウスを選択する。適格なウェアハウスが無い場合、WSSキューに残り続ける。ジョブスケジュールを行うと、各ウェアハウスのリソース使用状況バランスが変化する。 WSSはクラウドサービスにVM使用状況のレポートを送る。クラウドサービスは状況次第でDOPを下げる(より少ない並列度で処理するよう計画される)。 DOPを下げた後、ジョブはウェアハウスで実行される。ジョブ終了後リソースは解放される。負荷に応じてDOPがダイナミックに調整されている様が書かれている。実際のところ、DOPの推移を観察することはできないのと、DOPの上げ下げとパフォーマンスの関連が本当にその通りなのか不明なこともあり、結局良くわからない。並列レベルの制御 MAX_CONCURRENCY_LEVELパラメタにより、最大並列処理数を設定できる。デフォルト値は8ということなので、最大で4個のジョブを並列実行することになる。巨大なクエリを処理する場合、1個のジョブを受け持つコア数を増やすことでスループットが上がる場合があるらしい。並列処理数が下がるとキューに積まれるジョブが増えることに繋がる。ウェアハウスサイズを増やさずにMAX_CONCURRENCY_LEVELだけ調整しても、リソースの総量は変わらないはずだし、簡単に最適値が見つかるなら全自動で決めてくれるのだろうから、きっと難しい話なのだろう。QASみたいに全然違う何かを使うと良いよ、と書かれているこれは公式の以下のドキュメントが対応する。同時実行クエリの制限リソースモニタと使用量制限クレジットを想定よりも多く消費しないようにするアラートとハードリミットの仕組み。消費クレジットが制限を超えたことをトリガにアラート、自動停止を実行できる。リソースモニタが設定されていないウェアハウスを以下のクエリで見つけて設定せよとのこと。 SHOW WAREHOUSES ; SELECT \"name\" AS WAREHOUSE_NAME ,\"size\" AS WAREHOUSE_SIZE FROM TABLE(RESULT_SCAN(LAST_QUERY_ID())) WHERE \"resource_monitor\" = \'null\' ; ウェアハウスの負荷とサイズの決定方法 Snowsightでウェアハウスの負荷を確認できる。これの計算方法などが書かれている。確かに、あれ、何をどうやって集計したチャートなのか知らなかった。 Snowflakeが出力するメトリクスを見てウェアハウスの正しいサイズを決定せよとのこと。ウェアハウスのジョブ負荷メトリクスは、一定期間内の実行ジョブ数、キューに入ったジョブ数の平均である、とのこと。実行ジョブ数の平均は、全てのジョブの実行時間(秒)を期間(秒)で割った値であるとのこと。これはバーの青色の部分だな。 Private Previewで、ウェアハウスの使用率メトリクスが用意されるらしい。以下の表のように、ウェアハウス単位、クラスタ単位で100分率の値を得られる。ウェアハウス負荷や使用率によって、キャパシティ割り当てを行うべきとのこと。どういう数字だったらどうすべきか書かれている。そういえば知らなかった。ワークロードのスループット・レイテンシが適切で、キューに入ったクエリが少なく、長期にわたりクエリ負荷が1未満、かつ、使用率が50%を切る場合、ウェアハウス・クラスタのダウンサイズを検討する。別のウェアハウスを起動し、キューに入れられたジョブをそのウェアハウスで実行できるようにする。ワークロードのスループット・レイテンシが期待よりも低速で、かつ、クエリ負荷が低く、かつ、使用率が75%を超えるなど高い場合、ウェアハウスのアップサイズを検討するか、クラスタの追加を検討する。使用量の急増(スパイク)が繰り返し発生する場合、ウェアハウスの追加・クラスタの増量を行い、スパイクに対応するクエリをそれに移す。スパイク以外のクエリを小さいウェアハウス・クラスタで実行されるようにする。ワークロードが通常よりも大幅に高い場合、どのジョブが負荷に寄与しているのか調査する。ウェアハウスが定期的に実行される(スパイクではない)が、かなりの期間にわたって合計ジョブ負荷が1未満である場合、ウェアハウスのサイズダウン、クラスタの削減を検討する。ストレージ・キャッシュ-ストレージアーキテクチャ Snowflakeには、テーブルの永続化、JOIN等のクエリ演算子によって生成されクエリの実行中に消費される中間データの2つの形式のストレージがある。永続化テーブル寿命が長い永続化テーブルは、S3等のオブジェクトストレージが使われる。オブジェクトストレージは比較的スループットが高くないが、長期間保管する際の可用性要件が良い。 S3等のブロックストレージに対して一括上書きすることになるが、immutableなデータを扱うには適している。ブロックストレージの上でimmutableなデータの水平展開を行う。 (別のMedium記事で、micro-partitionはテーブルのバージョニングであって、immutableなデータ領域を重ねていくことと、その仕組みにより副作用的にTime-Travelが用意されることが書かれている。micro-partitionがブロックストレージ上で増えていく様は面白い) immutableなファイルには列データ、属性データがグルーピング・圧縮され格納されている。相対位置が付与されていて再構成しやすい。ブロックストレージに備わっている「部分的な読み取り」機能により、これらのファイルの必要な部分を取得する。こうして永続化テーブルがブロックストレージに保管・使用される。 JOIN等のクエリ演算子によって生成されクエリの実行中に消費される中間データ中間データは寿命が短く低レイテンシ・高スループットが求められる。ジョブの実行にウェアハウスのメインメモリとSSDが使われる。これらはウェアハウスの開始時に作られ、終了時に破棄される。これらの一時ストレージは、リモートにある永続化テーブルのライトスルーキャッシュとして機能する。各仮想ウェアハウスはそれぞれ個別に一時ストレージを持ち、クエリ実行時に使用される。この一時テーブルは、全ての仮想ウェアハウスから\"個別にコピーすること無しに\"共有できる。メモリ管理を単純化するためのSpill 中間データの書き込み操作の際に、まずウェアハウス内のメインメモリが使われる。メインメモリがfullになると、ウェアハウスのローカルdisk(SSD)が使われる。ローカルdiskがfullになると、リモートストレージが使われる。メモリ不足、ディスク不足を回避するための仕組みになっている。事実としては良く知られた挙動だけれども、それと「メモリ管理の単純化」というストーリーが紐づいて理解しやすくなった気がする。ストレージ・キャッシュ-キャッシュ戦略「キャッシュ」とは、良く使うデータを取り出しやすいところに一時的に保存しておくもの。キャッシュ容量は限られるため、ヒット率を維持しつつ効率的に中身を更新することが重要。その具体的な仕組みとして、LRU (Least Recently Used)、LFU (Least Frequently Used)が有名。キャッシュが必要な中間データ(前述)量が小さい場合、一時ストレージレイヤ(=VMのdisk)は、ファイル名のハッシュ値を使ったLRUキャッシュにより、頻繁にアクセスする永続化データのキャッシュとして使われる。このキャッシュは低優先度で\"lazy\"に行われるらしい。ファイルが仮想ウェアハウスのどのVMにストアされるかについて「一貫性」が言われている。一方向関数にファイル名を食わせた結果、ファイル名とストア先VMが決まることを言っている。サイズ変更によってVMの追加・削除が行われる際にキャッシュがシャッフルされてしまわない。 (VMのサイズが同じならば)永続化ストレージ上のファイルは特定のVMに保存されるため、永続化ストレージ上のファイルに対する操作は、そのファイルのハッシュが保存されるVMが実行するようにスケジューリングされる。こうして、ジョブの並列化はファイルのハッシュ値が一貫して同じVMに保存されることと密接に結びついている。ファイル名が偏っているとハッシュも偏り、保存先のVMが偏る場合がある。それを回避するため、ワークロードがそのVMでの所要時間が他のVMでの所要時間よりも小さいかどうか、に基づいてクラスタ内のVM内でロードバランシングが行われる。(え..?) キャッシュ(execution artifacts)が移動した場合(キャッシュアウトした場合)、最初に実行がスケジュールされていた既に過負荷になっているVMの負荷がさらに増加するのを避けるため、操作の実行に必要なファイルが永続化ストレージから読み取られる。仮想化の問題、ネットワークの問題など様々な理由で一部のVMが極端に遅い時があるらしい。その対策にもなっているらしい。 Snowflakeのスケジューリングロジックは、execution artifactsを永続化ストレージのキャッシュ先と同じVMに配置することと、全てのexecution artifactsを少数のVMに配置することの間のバランスを見つけようとする。前者は永続化ストレージのReadに伴うネットワークトラフィックの最小化を目指すが、ジョブがウェアハウス内の全てのVMにスケジューリングされることによって中間データが VM間でやり取りされることに起因してネットワークトラフィックが増加するリスクもある。後者は中間データ交換のためのネットワークトラフィックがなくなる(減る..?)が、永続化ストレージのReadのためのネットワークトラフィックが増加する可能性がある。一時データ容量はリモートの永続化ストレージ容量よりもかなり小さい(平均0.1%未満) にも関わらず、Snowflakeのキャッシュスキーム上では、Readのみのクエリで-80%、 Read-Writeがあるクエリで-60%のキャッシュヒット率にもなるらしい。文章だけでは読みきれないな..。ただキャッシュの仕組みが書かれているだけでなく、永続化ストレージ上のデータ(=ファイル)をVMに持ってくる仕組みの説明になっていて、ウェアハウス内のVMで負荷分散して処理していく様が薄ら分かった気がする。マルチテナント環境におけるセキュリティとリソース分離アカウント、ジョブごとにデータを分離し、アカウント、ジョブ間でデータが漏洩しないように設計している。\"仮想マシンを分離すること\"により、各テナントの分離を実現している。さらに、cgroup、カーネル名前空間、seccomp(※)のようなDockerコンテナに似たカーネルプリミティブを備えたVM内のサンドボックスにより、同一顧客アカウント内のジョブ間の情報漏洩を防ぐ。 ※cgroup,カーネル名前空間,secompはLinuxカーネルの機能で、 Dockerコンテナの内部で使われている。 cgroup,namespaceは、プロセスグループのリソース(CPU、メモリ、ディスクI/Oなど)の利用を制限・隔離するLinuxカーネルの機能とのこと。seccompは自プロセスが発行するシステムコールを制限してプロセスを乗っ取られたとしても被害を最小限にする機能とのこと。各VMを独自のハードウェア、ページテーブル、カーネルを使用して動作させることで、マルチテナントセキュリティとリソース分離を図っている。 VMが同じハードウェア、ページテーブル、カーネルを使用した\"VM分離\"がない場合、従来から使われているカーネルカーネル共有方式(cgroup,名前空間,secomp付き)だけでは、 Snowflakeのセキュリティ基準に達しないと判断したそう。(そうですか..)。 \"VM分離\"するよりもカーネルを共有した方が、コンテナは高速に起動して都合が良いけれども、カーネルを共有するということは、過去のCVEsから予想されるセキュリティ脆弱性に曝露されることになる。仮想ウェアハウスを構成するVMはそのウェアハウスが占有するプライベートなリソースであって、仮想ウェアハウス間で共有されたりはしない。加えて仮想ウェアハウスはステートレス。データの状態に影響されず、需要に応じてどんな時でも作成・破棄・リサイズできる。その仕組みのため、ジョブが特定の仮想ウェアハウスで限定して実行されるから、その仮想ウェアハウスのパフォーマンスが他の仮想ウェアハウスのパフォーマンスに影響しない。ジョブ実行の際、各仮想ウェアハウス内のVMが新しいプロセスを起動する。そのプロセスはジョブの実行期間中にのみ生存する。プロセスの失敗は自動的に検知され即座に修正(再実行)される。ユーザは、いつでも複数の仮想ウェアハウスを実行できる。各ウェアハウス上で、複数のジョブが並列実行する。ネットワークセキュリティ仮想ウェアハウスは次の外部ネットワークアクセスを必要とする。クラウドサービスレイヤとの通信ジョブ実行時に発生する他の仮想ウェアハウスとのデータ共有ローカルのクラウドストレージ(diskのspill先)へのアクセス API Gatewayへのアクセス Snowflakeは全ての仮想ウェアハウスからのネットワークトラフィックを信用しない。内部サービスへのトラフィックは必ず認証済みのエンドポイントを経由する。外部ネットワークへのトラフィックは外向きプロキシを経由し、アクセス制御ポリシーが適用される。未認証のエンドポイントへのアクセスはブロックされ、予期しない動きはSnowflakeに報告される。アカウント間で予期しない漏洩が起こらないように、VM、proxy、ジョブ間でやり取りされる全ての通信が正常であることを、クラウドサービスレイヤがIPアドレスマッチングを行うことで検証する。仮想ウェアハウスが持つ署名済みの共有シークレットを使って、仮想ウェアハウス間の全ての通信について、発信・着信側が本当にSnowflake内部の仮想ウェアハウスであるか検証する。そもそも仮想ウェアハウスからクラウドサービスレイヤへの通信がむちゃくちゃ多くなり、 DoS攻撃のようにならないように、通信にレートリミットがついていたりするらしい。他には、フローログを使って何かをしているらしい。フローログって何か知らなかったので調べた。 NWインターフェース間で行き来するIPトラフィックに関する情報をキャプチャする機能。とか。 Wireshakみたいなやつだろうか。例えば、仮想ウェアハウス内のVMが知らないdestに対して送ったIPトラフィックを見つけてforensic inspectionを行いVMを隔離するなど。 ※デジタルフォレンジック。「証拠保全」みたいな使われ方をしている。うーん..難しい... ネットワークセキュリティと言うと、つい外部から内部(Ingress)の事かなと思っていたが、 SaaSの内部で好き放題されてしまうリスクがある気持ちを理解した。外部ネットワークアクセスはこの気持ちの上に成立しているんだろう。 Python/Scala/Javaコードの分離 SQLみたいに出来ることが制限されている言語とは違い、何でもできるJava/Python/Scalaで UDFやプロシージャを書くことはセキュリティ面でリスクがいっぱい。これらの言語で書いた処理は、パフォーマンスの観点で、ジョブの他の処理と同じVM上で動く。マルチテナント環境上で(処理を?)分離するために(前述のように再利用できない)VMを使用するのに加え、cgroups, namespaces, secomp, eBPF, chrootのようなLinuxカーネルの要素を使ったセキュアなサンドボックスを提供することで、ジョブに割り当たったスコープの外の情報にアクセスしたり、処理がSnowflakeの他の機能に影響したりしないようにしている。 (これらは前述されている。それぞれうっすら調べてみた。こういう風に作るんだなぁと面白い) Java/Python/Scalaで書かれた各ジョブには、実行用に新たにサンドボックスが割り当てられる。コードの実行に最低限必要なread-onlyのソフトウエアが用意される。サンドボックス用のchrootが用意され(/より上に行けない)、その下には書き込み可能ディレクトリがいくつかあるだけ。ジョブはそこで処理を行う。read-onlyなディレクトリがマウントされて、 JavaのJARパッケージ、Pythonパッケージや、データファイルはそこで共有される。サンドボックス内のジョブ(のリソースを使用するプロセス)はcgroupが設定され、使用メモリ、CPU使用量、PID使用量(プロセス数?)が制限される。マルチプロセッサユースケース(マルチスレッド化してプロセス内で処理を並列化する話?)のためスレッド生成がサポートされる。さらに、許可リスト(IPC,Inter Process Communicationに関するリソースを隔離する仕組み= IPC Namespace、eBPF,extended Berkley Packet Filter=カーネル内で発生したイベントで駆動する処理を安全・簡単に組み込む仕組みによって、予め許可していないartifacts がサンドボックスの外に接続するUNIXソケットを開けないようにする)によるネットワークアクセスの制限、process namespaceによるVM上の他のプロセスを見えなくする制限、 seccomp(子プロセスのフォーク、実行可能プログラムの実行)によるカーネルAPIの不必要な実行の回避が行われる。脅威検知のためptraceがシステムコールを管理する。ジョブが完了した後、VM上の環境のもろもろの解放、開いたソケットのクローズ、クレデンシャルの削除、ローカルキャッシュ、一時ファイル、ログの削除が行われる。追加の多層防御手段?(defense-in-depth measure?)として、規定時間内に終了しなかった Python/Job/Scalaコードを実行するプロセスに対して、監視プロセスがkillシグナルを送る。サンドボックス外に離脱したり、攻撃者が仮想ウェアハウス上のVMにプロセスを残したりルートキットを配置する未知のリスクに備えて、Python/Java/Scalaコードを実行したVMは「実行不可」としてマークされる。仮想ウェアハウスのスケジューリングや起動済みVMをプールする仕組みの上で、Python/Java/Scalaコードを実行したVMが異なるアカウント・ユーザに割り当てられると、アカウント間情報漏洩のリスクに繋がってしまうため、異なるアカウント・ユーザに割り当たらないようになっている。Python/Java/Scalaコードを実行するVMが作られると、アカウント専用のVMプール入れられる。新しいVMを割り当てるときは、まずはアカウント毎の空きプールからVMが選ばれる。多数のゼロデイエクスプロイト（脆弱性が発見されてからパッチが当たるまでの期間の攻撃)が連続して使用されると、サンドボックスが破られてしまうかもしれないが、それに備えた作りになっている。まずエクスプロイトは、ユーザアカウントで実行中のVMに存在する。このVMは、 Snowflakeサービスや、Snowflake内のローカルネットワーク上のVMから隔離されている。攻撃者が手にしたクレデンシャルは(サンドボックスを破壊した)特定のアカウントの特定のVMに限定され他では使用できない。あくまで論理的な構成が書かれているだけで「コンテナ」というワードも無いし、何かチラチラとするな。こういうのを「コンテナエスケープ」とか言うらしい。ソフトウエア更新の管理 Snowflakeの各機能がどうやって仮想ウェアハウスにデプロイされるかについて。 (デプロイの)ワークフローにより新機能、セキュリティアップデート、機能改善が行われる。全ての処理は自動化されていて手作業の間違いが起きないようにしている。このリリースプロセスにおいて、単体テスト、回帰テスト、結合テスト、性能、負荷テストが行われる。リリースプロセスは、本番の前段の環境、または本番に近い環境で行われる。 VMがフリープールに入る前に最新のパッチが当たる。VMのStartやResumeなどの操作の後に、フリープールからVMに割り当たったり、逆にVMからフリープールに抜けたりするが、フリープールからVMに割り当たるプロセスの一部として、VMに最新に保つための最新のバイナリがダウンロードされ、適用される。 Resume、Startなどのライフサイクル操作は即座に終わるように作られているが、影響を与えないように性能要件が与えられているらしい。 SKU sizeやOSのメジャーパーションなど大きな変更の際には、未適用のVMと適用済みのVMの両方が同時に動く状態となる。古い方は既存のジョブを実行し、新しい方は、新しいジョブを実行する。そのようにジョブがルーティングされる。既存のジョブを実行し終わってから、最終的に古い方は消される。つまり、1個のウェアハウスについて、アップデートの時期を迎えると背後で(適用前後の)2個になる。前述のようにキャッシュはVMのローカルディスクなので、もし古いウェアハウスが破棄されたとすると、キャッシュが失われることになる。それによりキャッシュミスが発生しパフォーマンスに影響しないように、事前に管理されているとのこと。がんばってテストしているけれども運用環境にバグが混入することもある。なのでアップデートをロールバックできるようになっている。クラウドプロバイダのリージョン毎に、動作中のバイナリの背後で、古いバイナリをコピーしている。古い方は非アクティブのままとしている。(トラフィックが発生しない?) 大規模障害に備えて、通常、新しいジョブを新しいバージョンのウェアハウスにルーティングしているものを古いウェアハウスにルーティングするロールバックをできるようにしている。 Issueに基づいて顧客ごとに対象を絞ったロールバックをすることもあるらしい。顧客のワークロードはそれぞれ大分ことなるので、全員が同じ頻度でバグを踏むことはないので。特定の顧客に対して、アップデートした一部のリリースをロールバックする、みたいなことをするらしい。リリースノートの扱いが良い感じになっていて、こういう感じで運用されているのだな、と。将来の機能現在、ユーザは、ワークロードの複雑さ、処理時間、コストを考慮して適切な調整を行わないといけない。例えば、サイズ、ウェアハウスタイプ、クラスタ数、スケーリングポリシーなど。こういったキャパシティ調整の大変さを減らしたり無くそうとしているらしい。 microVM(例えばFirecrackerやKata Containersなど)やシステムコールのオフロードに投資し、より強力なサンドボックス分離メカニズムを実現しようとしているらしい。それにより、Python/Javaコードで現状ではできないことが出来るようにしたいらしい。まとめ Deep dive into the internals of Snowflake Virtual Warehousesを読んでみました。たぶん公開されていない内部の仕組みの割合が多いのかなと思いましたがどうでしょうか。正直かなり難しくて、途中、ほとんど写経状態になっている部分もありますが、なるべく分からないところを調べながら、何を言いたいのかを趣旨の理解に努めました。正直、知らなくても問題ないし、公開されていない以上、実際は違うかもしれないし、将来変更されてしまうかもしれません。 1週間ぐらいかけて読んでみて、公開されている仕様を説明しやすくなった気はしました。