この記事は、Luup Advent Calendar の 22日目の記事です。
こんにちは、Data Scienceチームの長谷川（@chase0213）です。
Data Scienceチームでは、社内の様々な部署からデータ分析に関する依頼を受けたり、自ら課題を見つけ仮説検証したりして、単純な集計から複雑なモデリングまでデータにまつわることを幅広く行っています。
分析用のデータは基本的にデータウェアハウス（BigQuery）に格納されており（参考: 冪等性を担保したGoogle Cloud Composerの設計と実装）、単純な集計や日常的な指標の監視には、可視化や共有のしやすさから Redash を利用しています。
一方で、少し複雑な分析をする場合には、自由度の観点から jupyter や Google Colaboratory などの別環境にデータを移して分析することが多いです。
そこで問題になってくるのが、分析環境にデータを移すことにより生じる、分析環境とデータウェアハウスとの間のデータ同期ズレの問題です。
ごく少量のデータが異なっているくらいであれば誤差とみなせることも多いですが、実際にはデータウェアハウス（BigQuery）内のテーブルのスキーマや集計ロジックは日々変更されることがありますし、それによって過去実行できていたクエリが実行できなくなるということも起こり得ます。
法律やマーケットなど、自社でコントロールできないことが多い Luup の事業においては、あらかじめ全てのデータの変更可能性を予見するのは不可能であり、分析者として受け入れなければならない試練です。
今回の記事では、この問題について行っている対策の1つをご紹介します。
Pandera が解決する課題
Pandera は、pandas.DataFrame のような dataframe形式のデータをテストするためのライブラリです。
例えば、次のように pandas.DataFrame形式のデータがあるとします。
import pandas as pd
# カラムが 3列ある適当なデータを作成
df_original = pd.DataFrame({
"price": [50, 80, 0, 110, 200],
"ride_duration_minutes": [10.2, 21.3, 8.8, 9.1, 15.2],
"user_id": [123, 234, 345, 456, 567],
})
人間はこのデータを見て「price は整数で、おそらく10の倍数を期待しているんだな」「ride_duration_minutes は浮動小数点型かな。カラム名的に負数は取らなさそうだな」「user_id は整数型だな。もしかしたら後で変わるかも。」と判断できます。
これを python では以下のように判定しており、期待通りに認識していることがわかります（Dtype のカラムを見ています）。
>>> df_original.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 5 entries, 0 to 4
Data columns (total 3 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 price 5 non-null int64
1 ride_duration_minutes 5 non-null float64
2 user_id 5 non-null int64
dtypes: float64(1), int64(2)
memory usage: 248.0 bytes
さて、では次のようなデータではどうでしょう？
df_changed = pd.DataFrame({
"price": [50, 80, 0, 117, 200],
"ride_duration_minutes": [10.2, 21.3, 8.8, 9.1, 15.2],
"user_id": [123, 234, 345, 456, "u_567"],
})
先ほどとの違いで言うと、priceカラムの 110 という値が 117 に、user_idカラムの 567 という値が "u_567" に変わっています。
これをどう python が認識するかというと、以下のようになります。
>>> df_changed.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 5 entries, 0 to 4
Data columns (total 3 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 price 5 non-null int64
1 ride_duration_minutes 5 non-null float64
2 user_id 5 non-null object
dtypes: float64(1), int64(1), object(1)
memory usage: 248.0+ bytes
price は変わらず int型、user_id は object型（string型は pandas では object型になります）と判断されました。
python は静的型付け言語ではないため、データの変更があってもエラー無く読み込めてしまうことがメリットであり、デメリットでもあります。
ことデータ分析の文脈で言えば、これによって分析・集計結果が「壊れる」こともあり、一概に好ましい性質とは言えません。
例えば、ユーザーを無作為抽出してなにか検証したい場合に以下のようなコードを書くことがあります。
df_sampled = df_changed[df_changed.user_id % 100 == 0]
こういった場面では、型が変わってしまうとエラーになったり、最悪の場合 python側でよしなに解釈されてエラーにはならないが集計結果は壊れている、ということが起こり得ます。
このようなデータの変化は例えば料金形態の変更や ID の採番ロジックの変更などで往々にして起こりえます。
ここで問題なのは、このデータの変化がビジネス的にも、データ基盤的にも正しいものということです。
何らかの原因でデータにノイズや欠損が生じたのであれば、データウェアハウス（データ基盤）側で検知・整備できますが、今回のケースでは機械的に対応できません。
また、仮にデータウェアハウス側で整備が十分に行われていたとしても、分析結果の健全性を保つ意味では、分析側でもチェックする方が望ましいです。
そこで、Pandera が提供している DataFrame の型や値チェックの機構を使用して、このような望ましくない挙動を事前に防ぐことを考えます。
Pandera の基本
前述のような DataFrame に対して、Pandera では Schema を定義してチェックを行います。
import pendera as pa
# schema定義
schema = pa.DataFrameSchema({
"price": pa.Column(int, checks=[
pa.Check(lambda p: p % 10 == 0)
]),
"ride_duration_minutes": pa.Column(float, checks=pa.Check.ge(0.0)),
"user_id": pa.Column(int, checks=pa.Check.ge(0)),
})
簡単に説明すると、次の部分で priceカラムが int型であること、そしてその値が 10の倍数あることを定義しています。
"price": pa.Column(int, checks=[
pa.Check(lambda p: p % 10 == 0)
]),
同様に、次では ride_duration_minutesカラムが float型であることと、値が 0.0以上であることを定義します。
"ride_duration_minutes": pa.Column(float, checks=pa.Check.ge(0.0)),
そして、以下の部分では user_idカラムが int型であり、0以上の値であることを示しています。
"user_id": pa.Column(int, checks=pa.Check.ge(0)),
ここまではあくまで schema を定義した段階であり、チェックが実行されるのは次のように schema に DataFrame を渡したタイミングです。
>>> schema(df_original)
price ride_duration_minutes user_id
0 50 10.2 123
1 80 21.3 234
2 0 8.8 345
3 110 9.1 456
4 200 15.2 567
値が定義通りの場合、schema() はなんのエラーも発出せず、DataFrame そのものを返します。
一方で、df_changed を入力すると、以下のように例外を発出します（実行ユーザー名などの情報は xxxxx で伏せています）。
>>> schema(df_changed)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 804, in __call__
return self.validate(
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 513, in validate
return self._validate(
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 710, in _validate
error_handler.collect_error("schema_component_check", err)
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/error_handlers.py", line 32, in collect_error
raise schema_error from original_exc
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 702, in _validate
result = schema_component(
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 2044, in __call__
return self.validate(
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schema_components.py", line 214, in validate
validate_column(check_obj, column_name)
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schema_components.py", line 187, in validate_column
super(Column, copy(self).set_name(column_name)).validate(
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 2002, in validate
error_handler.collect_error("dataframe_check", err)
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/error_handlers.py", line 32, in collect_error
raise schema_error from original_exc
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 1997, in validate
_handle_check_results(
File "/home/xxxxx/.cache/pypoetry/virtualenvs/xxxxx/lib/python3.8/site-packages/pandera/schemas.py", line 2383, in _handle_check_results
raise errors.SchemaError(
pandera.errors.SchemaError: <Schema Column(name=price, type=DataType(int64))> failed element-wise validator 0:
<Check <lambda>>
failure cases:
index failure_case
0 3 117
エラーの内容を見ると、pandera.errors.SchemaError: <Schema Column(name=price, type=DataType(int64))> failed element-wise validator 0: と書かれており、
priceカラムが期待したチェックロジックを満たさなかったこと（10の倍数ではなかったこと）を知らせています。
user_id も期待通りではない値が入っていますが、先に raise されたエラーしか表示されない点に注意が必要です。
実際にどういった場面で価値を発揮する（した）のか
Pandera がどういったものなのかご理解いただけたところで、実際にこれがどのような場面で役に立つのか考えてみます。
そもそも、データ分析を BigQuery ではなく python でやらなければならないケースとしては、データに対してより複雑な変換をし、その結果を利用する場面が大多数なのではないでしょうか。
そして、python（もしくは他のプログラミング言語）で行わなければならない複雑な変換は、その変換処理自体の計算量も時間複雑性、空間複雑性ともに大きいかもしれません。
すると、大量のデータをデータウェアハウスから取ってきて重い処理を行い、しばらく待った後に値の不整合によるエラーが出て全て最初からやり直し……ということが起こります。
もし、データウェアハウスの構造やデータ定義が未来永劫変わらなければ、一度作った分析用のスクリプトもメンテナンス不要で価値を発揮し続けます。
しかし大抵の場合、そんな都合の良いことはなく「このスクリプトはあまりに複雑な処理を行っているので、最初から作り直した方が早い」ということも起こり得ます。
それを回避するためには、何かデータに変更があればそれをいち早く検知して、小さく更新を重ねることが重要となります。
さらに、分析スクリプトを作った人が将来に渡ってその分析スクリプトをメンテナンスするとは限りません。
pandas.DataFrame は非常に柔軟で強力なフレームワークである一方、各カラムが何を示しているか、どのカラムが存在すべきなのか、どのカラムは不要なのかといった情報は保持されません。
1つの解決策はコメントを残すことですが、多くの分析はトライ＆エラーを繰り返し非常に短いスパンで頻繁にコードを書き換える必要があるため、ある程度形になったらコメントをアップデートするという方針で分析する人も多いのではないでしょうか？
もしその際にコメントのアップデートを忘れていたとしても、コードがコメントに反しているか否かに依らずコードは動いてしまうため、もし仮にレビュープロセスでも拾いきれなければ、整合性のないコードとコメントが誕生してしまうことになります。
そこで、分析前のデータチェックによってデータの性質が変わっていないことが保証されるのであれば、その分析スクリプトのメンテナンス性は極めて高くなり、分析結果を健全に保つことに貢献します。
まとめ
今回は、Pandera を用いてデータの整合性チェックを行う手法と、それが役に立つ場面をご紹介しました。
データサイエンスは、分析によって得られた結果を持ちいて利益向上に貢献する「攻め」の側面に光が当たることが多いですが、一度作った分析フローをメンテナンスしやすくする「守り」も考えていく必要があります。
仮説検証のプロセスを心地よく行うために、想定するデータの正しさに疑いの目を向け、分析結果の健全性を保つことを試みてみてはいかがでしょうか？
終わりに
Luup でのデータ基盤構築や、データ活用、データドリブンな戦略策定に少しでもご興味がある方は、まずは情報交換という形でも良いのでお話ししてみませんか？
この記事を読んでいただいているあなたとお話できる日を心待ちにしてます。
Discussion