Pythonでアドホックな分析､あるいは探索的データ分析(EDA)をやっていく過程で､
「あれ､あの時の結果(出力)って､どうやって出したんだっけ…？」
という状況になることはありませんか？ 筆者はあります｡
データ分析をするときに､どんなことを試したか自動で記録されたらいいのに…
と思ったので､それを実現する方法を記事にします｡
結論からいうと､いい感じにloggingとデコレータを使おうという内容です｡
なお､この記事の中でloggingの詳細についての解説は行いません｡ 1
この記事の想定読者
- Python(特にJupyter)でデータ分析をすることがある
- 投入する変数をいろいろ変えて試行錯誤することがある
- 一度やった分析の条件をあとから確認できるようにしたい
- データ分析の再現性を高めたい
この記事の成果物のサンプル
ある広告掲載に関する(架空の)データについて､
いろいろ条件を変えて結果を調べたい､という状況を考えます｡
以下のコードを実行すれば､変数にどんな値を代入して､
結果としてどんな値が返ってきたかを自動的に記録することができます｡
import pandas as pd
from funclogger import set_logger, funclog # この記事で作成するモジュール
logger = set_logger() # ここでloggerをセット
df = pd.read_csv('sample_data.csv')
@funclog(logger) # 記録したい関数にデコレータを適用する
def diff_max_min(df: pd.DataFrame, col: str, name: str, num_col: str):
"""特定の条件に一致するデータについて､最大値と最小値の差を計算する"""
df = df[df[col]==name]
result = max(df[num_col]) - min(df[num_col])
return result
# あとは普通に関数を実行する
diff_max_min(name='東京新宿', col='広告掲載場所', num_col='閲覧者数', df=df)
diff_max_min(name='福岡天神', col='広告掲載場所', num_col='閲覧者数', df=df)
結果として出力されるファイルがこちらです｡
2022-04-16 20:09:44,222,DEBUG,diff_max_min,args=[],kwargs={'name': '東京新宿', 'col': '広告掲載場所', 'num_col': '閲覧者数', 'df': ' 日付 広告掲載場所 閲覧者数 0 2022-04-27 仙台 '}
2022-04-16 20:09:44,223,DEBUG,diff_max_min,return=4210
2022-04-16 20:09:44,225,DEBUG,diff_max_min,args=[],kwargs={'name': '福岡天神', 'col': '広告掲載場所', 'num_col': '閲覧者数', 'df': ' 日付 広告掲載場所 閲覧者数 0 2022-04-27 仙台 '}
2022-04-16 20:09:44,225,DEBUG,diff_max_min,return=6278
ちょっと長く感じますが､中身は以下のような構造になっています｡
{実行された時間},{ログの重要度レベル},{実行した関数名},{位置引数args},{キーワード引数kwargs} # 文字列が長くなる場合は適当な位置でカット
{実行された時間},{ログの重要度レベル},{実行結果}
位置変数argsとキーワード引数kwargsにどんな値が代入されたのか､
そして結果としてどんな値がreturnされたかが記録されています｡
今回はそこまで複雑なケースではないですが､やっていることが複雑になるほど､
試行回数が増えるほどに心強いバックアップになってくれる…はずです｡
ただし､実行した関数の中身が変更されたことは感知できません｡
ちなみにsample_dataの中身はこんな感じです｡
中身は本当にランダムに生成したので､実際の事象とは一切関係ありません｡2
日付,広告掲載場所,閲覧者数
2022-04-27,仙台,9209
2022-04-28,仙台,3331
2022-04-18,札幌,6980
2022-04-20,埼玉大宮,9798
2022-04-18,仙台,5933
(…以下同様にデータが続く)
コードの内容
コード全文
いきなりコード全文を載せます｡
中身は順番にざっくりと解説していきます｡
import re
import os
import functools
import logging
import sys
LOG_FORMAT = "%(asctime)s,%(levelname)s,%(message)s"
STREAM_LEVEL = logging.INFO
FILE_LEVEL = logging.DEBUG
SAVE_FOLDER = 'log'
_main = os.path.abspath(sys.modules['__main__'].__file__)
MAIN_NAME = os.path.basename(_main).rstrip('.py')
def set_logger(module_name=MAIN_NAME):
if not os.path.exists(SAVE_FOLDER):
os.mkdir(SAVE_FOLDER)
save_path = f'{SAVE_FOLDER}/{MAIN_NAME}.log'
logger = logging.getLogger(module_name)
logger.handlers.clear()
streamHandler = logging.StreamHandler()
fileHandler = logging.handlers.RotatingFileHandler(
save_path, maxBytes=1000000, backupCount=5)
formatter = logging.Formatter(LOG_FORMAT)
streamHandler.setFormatter(formatter)
fileHandler.setFormatter(formatter)
logger.setLevel(logging.DEBUG)
streamHandler.setLevel(STREAM_LEVEL)
fileHandler.setLevel(FILE_LEVEL)
logger.addHandler(streamHandler)
logger.addHandler(fileHandler)
return logger
def funclog(logger, arg_maxprint=50, ret_maxprint=100):
"""__main__で生成したloggerを代入して、関数のログを取る"""
def _funclog(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
args_str = [re.sub('[\s,]', ' ', str(arg))[:arg_maxprint] for arg in args]
kwargs_str = {k:re.sub('[\s,]', ' ', str(v))[:arg_maxprint] for k, v in kwargs.items()}
logger.debug(f"{func.__name__},args={args_str},kwargs={kwargs_str}")
ret = func(*args, **kwargs)
ret_str = re.sub('[\s,]', ' ', str(ret))[:ret_maxprint]
logger.debug(f"{func.__name__},return={ret_str}")
return ret
return wrapper
return _funclog
def deco_funclog(func, logger):
"""関数にfunclogのデコレータをつける
ただし､デコレータを重複させない
"""
if '__wrapped__' in dir(func):
return func
@funclog(logger)
@functools.wraps(func)
def wrapper(*args, **kwargs):
ret = func(*args, **kwargs)
return ret
return wrapper
冒頭の処理
最初にログのフォーマットやレベルを設定しています｡ 3
その次にログファイル保存用に､現在実行しているPythonファイルの名前の取得をしています｡4
定数化している部分はset_logger()を実行する前であれば差し替え可能です｡
LOG_FORMAT = "%(asctime)s,%(levelname)s,%(message)s"
STREAM_LEVEL = logging.INFO
FILE_LEVEL = logging.DEBUG
SAVE_FOLDER = 'log'
_main = os.path.abspath(sys.modules['__main__'].__file__)
MAIN_NAME = os.path.basename(_main).rstrip('.py')
set_logger()
set_logger()の部分はryoheiszkさんのこちらの記事の内容を参考にさせていただきました｡
ファイルの保存や､一部の定数化に関する部分だけ変更しています｡
def set_logger(module_name=MAIN_NAME):
if not os.path.exists(SAVE_FOLDER):
os.mkdir(SAVE_FOLDER)
save_path = f'{SAVE_FOLDER}/{MAIN_NAME}.log'
logger = logging.getLogger(module_name)
logger.handlers.clear()
streamHandler = logging.StreamHandler()
fileHandler = logging.handlers.RotatingFileHandler(
save_path, maxBytes=1000000, backupCount=5)
formatter = logging.Formatter(LOG_FORMAT)
streamHandler.setFormatter(formatter)
fileHandler.setFormatter(formatter)
logger.setLevel(logging.DEBUG)
streamHandler.setLevel(STREAM_LEVEL)
fileHandler.setLevel(FILE_LEVEL)
logger.addHandler(streamHandler)
logger.addHandler(fileHandler)
return logger
このloggerだけでもloggingは可能ですが､
今回は更にものぐさにログを残せるようにデコレータを準備しています｡
funclog()
定義する関数にデコレータとして適用すると､
投入した変数と返り値の両方を自動でログに残してくれます｡
中身は3層の入れ構造になっているので少しややこしいです｡
かいつまんで説明すると､loggerと対象の関数(func)を受け取って､
その変数(args,kwargs)と返り値(ret)を文字列にしてログとして出力しています｡
途中のre.sub()やmax_printは出力する文字列の改行をなくしたり､
極端に文字列が多くなりすぎないように制御をする役割を果たしています｡
def funclog(logger, arg_maxprint=50, ret_maxprint=100):
"""__main__で生成したloggerを代入して、関数のログを取る"""
def _funclog(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
args_str = [re.sub('[\s,]', ' ', str(arg))[:arg_maxprint] for arg in args]
kwargs_str = {k:re.sub('[\s,]', ' ', str(v))[:arg_maxprint] for k, v in kwargs.items()}
logger.debug(f"{func.__name__},args={args_str},kwargs={kwargs_str}")
ret = func(*args, **kwargs)
ret_str = re.sub('[\s,]', ' ', str(ret))[:ret_maxprint]
logger.debug(f"{func.__name__},return={ret_str}")
return ret
return wrapper
return _funclog
deco_funclog()
最後に､定義後の関数にfunclog()を適用するために､
deco_funclog()という関数を準備しています｡
__wrapped__の部分は､関数を重複してデコレートしないようにする回避策です｡
Jupyterで繰り返し同じセルを実行していると､
ログも重複して出力されることがあるので念のため｡
def deco_funclog(func, logger):
"""関数にfunclogのデコレータをつける
ただし､デコレータを重複させない
"""
if '__wrapped__' in dir(func):
return func
@funclog(logger)
@functools.wraps(func)
def wrapper(*args, **kwargs):
ret = func(*args, **kwargs)
return ret
return wrapper
これはたとえば､別のモジュールで定義された関数を利用する場合などに使います｡
冒頭の例と同じ関数を別モジュールに定義します｡
import pandas as pd
def diff_max_min(df: pd.DataFrame, col: str, name: str, num_col: str):
df = df[df[col]==name]
result = max(df[num_col]) - min(df[num_col])
return result
これをメインのモジュールから呼び出します｡
そうするとそのままではデコレータを適用できないので､
deco_funclog()で関数の上書きをします｡
import pandas as pd
import sub_calc
from funclogger import set_logger, deco_funclog ,funclog
logger = set_logger()
df = pd.read_csv('sample_data.csv')
sub_calc.diff_max_min = deco_funclog(sub_calc.diff_max_min, logger)
sub_calc.diff_max_min(name='仙台', col='広告掲載場所', num_col='広告閲覧数', df=df)
sub_calc.diff_max_min(name='埼玉大宮', col='広告掲載場所', num_col='広告閲覧数', df=df)
2022-04-16 20:42:53,051,DEBUG,diff_max_min,args=[],kwargs={'name': '仙台', 'col': '広告掲載場所', 'num_col': '閲覧者数', 'df': ' 日付 広告掲載場所 閲覧者数 0 2022-04-27 仙台 '}
2022-04-16 20:42:53,052,DEBUG,diff_max_min,return=6532
2022-04-16 20:42:53,053,DEBUG,diff_max_min,args=[],kwargs={'name': '埼玉大宮', 'col': '広告掲載場所', 'num_col': '閲覧者数', 'df': ' 日付 広告掲載場所 閲覧者数 0 2022-04-27 仙台 '}
2022-04-16 20:42:53,053,DEBUG,diff_max_min,return=6283
まとめ
- 細かなloggingはlogger.info()などで設定する
- 変数と返り値だけならfunclog()でデコレートする
- 定義済みの関数に対してはdeco_funclog()を使う
これで分析の再現性が少し高まる…はずです｡
-
loggingの詳しい仕組みについては､こちらのページがおすすめです｡
【Python】仕組みを理解してログ出力を使いこなす ↩ -
ランダムなサンプルデータを作成する方法について､別の記事にまとめています｡
Pythonを使ってデータ分析のデモ用にそれっぽいCSVファイルを生成する ↩ -
ここでは
LOG_FORMATの属性をシンプルなものにしています｡
他に使える属性については､公式ドキュメントをご参照ください｡
logging --- Python 用ロギング機能 LogRecord 属性
また､出力例についてはシラベルノートさんのこちらのページがおすすめです｡
【Python】logging フォーマットの出力例（１０種類以上） ↩ -
この内容について別の記事を書いています｡
書き出すファイルの名前に､書き出しを実行したPythonファイルの名前を自動で挿入する方法【自作モジュール編】 ↩