はじめに

Python Advent Calendar 2020 25日目です。

時代はサーバレースですが、オンプレミスのサーバやクラウド（IaaS）などのインスタンスでバッチを運用しているシステムもまだたくさんあると思います。

本記事ではサーバ上で動かすバッチを前提とし、前半はアンチパターンなどを踏まえながらバッチ設計のポイントについてまとめています。後半はPythonでバッチ開発する際のTipsになります。

※本記事の内容は、あくまで考え方の一例であり、必ずしも全ての考え方がシステムに適合したり、ここに書いている内容で満たされている訳ではありません。

バッチ設計

バッチ処理はひとまとまりのデータに対して、一連の処理を連続で実行する処理方式のことです。語源を辿ると、汎用コンピューターの時代まで遡ります。

データを一括で処理することを目的とし、Unix系OSではcronを利用して指定した日時で運用することが多いです。また、バッチ自体のことをジョブと呼んだりもします。大規模システムではたくさんのジョブを管理するため、専用のジョブ管理サーバを構築し、ジョブ管理ソフトを導入して管理を行ないます。

極論、システムによってはバッチ処理を行わなくても手動で運用することはできます。しかし、現実的には人手のコスト、処理時間、確実性などを満たすために、バッチ処理は欠かせません。また、システム開発時に要件定義工程では必要性が分からなかったが、運用に入ってから必要になってくるケースもあります。

よって、バッチ設計はシステム全体を俯瞰しながら、以下に記載する様な基本的なポイントや、アンチパターンなど運用を考慮した作り込みがとても重要です。

基本的なポイント

バッチ内で使用する変数は保守性を高めるために、分かりやすい変数名をつける（xやiなどを付けない）
メソッドはシンプルにして、複数の処理をまとめない
環境毎に変わるようなDB接続の設定などは、configなどにして実行ファイルと分離する
ログモジュール等汎用的に使用する処理は、必要に応じてutilにして作成する
cron等ジョブ管理を行う場合は、突き抜けを考慮した設計が重要
DBのレコードに登録や更新を行う場合は、created_atやupdated_atのカラムを作成する
トランザクション実行時のコミット件数（データ量に応じて、スループットやロールバックを考慮）
リランを考慮した作り込み（リカバリー方式をシンプルにする）

アンチパターン

突然、システムの運用を引き継ぐごとで経験した、バッチ処理に関わるアンチパターンを以下に記載します。

ログローテーションされないログファイル
Pythonで作成されたバッチで、loggingモジュールを使用してログ出力されている。しかし、プログラム側でログローテーションを行っていないため、同じログファイルにログ出力が行われ続けている。

Linuxのrsyslogを知っていれば、プログラムで実装しなくてもOSの設定だけで解決できます。

複数作成されているcongig
以下のようにa,b,cと各バッチが格納されているディレクトリ構成になっている。各バッチはそれぞれ目的が違うが、データ連携を行うためのDB設定などは共通である。また、各configにはプログラムエラーなど例外発生時の通知先であるWebhookのURLも記載しているが、全て同じである。

.
|-- a
|   `-- config
|-- b
|   `-- config
`-- c
    `-- config

例えば、サーバ移行などが発生し、WebhookのURLを変更する際は全て書き換えないといけません。

バッチ処理の突き抜け
バッチ処理の中でインスタンスの起動・停止を行っているバッチがある。ある日、AWS全体に負荷がかかり、その影響を受けているせいか、普段よりインスタンスの起動・停止処理に時間がかっていた。そのため、先行のバッチ処理の時間が長くなり、後発のバッチ処理と重なり、バッチ処理が失敗。

想定外のシステム異常などによるバッチ処理の突き抜けを考えていない、バッチ設計の考慮不足。

いつ更新が行われたか分からないDB
バッチ処理で更新が行われたDBを調査することになったが、更新時間を記録していないため、調査が難航した。バッチ処理でDBの登録や更新を行う場合は、DB設計としてcreated_atやupdated_at等のカラムを作成すると、障害発生時などに調査がしやすくなる。

データがいつ登録・更新されたのか分からない場合は、運用・保守性を大いに低下させる。

何でも通知するバッチ
確認する必要がない事象に対するinfoメッセージの通知や、一時的に通信が失敗し接続ができなくて全てerror通知する様なバッチは、運用を引き継ぎいだ人にとっては意味がない。また、不要なerror通知は形骸化するとただのオオカミ少年になるため、システム運用として有害である。

サービスに影響があるエラーや、運用者に確認が必要なメッセージ以外は通知を行わない。

拡張性を考慮していないバッチ
バッチ処理の時間がサービスに影響するが、運用後バッチ処理の量が増える場合、並行処理や拡張性を考慮していないと順次処理に限界がある。

チューニングで対応できるのであれば問題ないが、拡張性を考慮していない設計は、後々の影響が大きくなる。

pipで不要なライブラリをインストール
サーバ移行が発生し、移行元サーバでpip freezeを実行して、requirements.txtを作成。新しいサーバでrequirements.txtを元にインストールすると、使用していないライブラリに関するエラー発生。

不要なライブラリはインストールしない。

ドキュメント

引き継ぎなどでシステムを担当する運用者が変わる場合、継続的に運用を行うためには、俗人化されたシステムほどドキュメントの整備が重要です。

システムで何らかの異常発生時に、運用者に異常を伝えるためにアラートを通知する設定がされているとします。突然、運用を引き継いだときに、アラートが適切に設定されてない場合は、障害すらも分からないため、infoかerrorの切り分けから行ないます。

仮にerrorだとして、バッチのログファイルを探します。
しかし、どこに出力されているのか分からないため、情報を手がかりに調査を行ないます。また、恐ろしいのはバッチのログ出力すら行われてない場合もあります。

システムの運用担当が変わる場合は、システム全体のバッチスケジュールや、バッチ一覧等、最低限システム全体が分かるドキュメントがあることが望ましいです。特にデータ連携を多く行うシステムほど、注意が必要です。

開発スタイル

Pythonの開発方法は色々あると思いますが、以下は開発効率を上げるための一例になります。

ローカル環境のDockerでPythonイメージを作成し、ソースを格納しているディレクトリをマウントすると、開発効率が上がります。

はじめにDokcerfileを作成してビルドします。
次にソースを格納したディレクトリをマウントした状態で、docker runコマンドで起動します。

Dokcerfileの作成

FROM python:3

WORKDIR /usr/src/app

COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD [ "python", "./your-daemon-or-script.py" ]

ビルド
$ docker build -t python3/test .
コンテナの起動
$ docker run -v ＜ソースディレクトリ＞:/batch -it python3/test /bin/bash

以降、エディタでソースディレクトリに格納したプログラムファイルの開発ができます。

Tips

Pythonでバッチ開発する際のTipsになります。

config

開発環境や本番環境など複数の環境が存在する場合は、config.pyなどのファイルを作成し、インポートすれば煩雑になるのを防ぐことができます。

sys.argvはPythonスクリプトに渡されたコマンドライン引数のリストです。argv[0]はスクリプトの名前で、argv[1]に1個目の引数が入ります。

例えばsys.argvを使用して、以下の様に環境毎の設定が行えます。
sys.argvの値を見て引数チェックを行う使い方は多いと思います。

import sys

args = sys.argv
env = args[1]

if env == 'local':
    pass

ログファイル

ログ出力を行うバッチファイルを作成したが、ログファイルが存在しないためエラーになることあります。

所定のログファイルが存在しない場合は、ログファイルを作成する処理を入れることで防げます。

log_file = config.base_dir + 'log/bacth.log'
if not os.path.exists(log_file):
    with open(log_file, 'w') as f:
        f.write('')

除外リスト

除外リスト（exclude_list）になければ、処理対象のリストに追加

if item_id not in exclude_list:
    stock_list.append({"item_id": item_id})

辞書で取り出して、リストに格納

DBの結果（result_set）を辞書で取り出して、リストに追加

for row in result_set:
    row_dict = {"id": row[0], "name": row[1], "age": row[2]}
    target_list.append(row_dict)

複数のリストから取り出して、リストに追加

リストだとずれる場合があるので、基本的には辞書を使うのがいいと思います。

for (z, x, y) in zip(list1, list2, list3):
    temp_list.append([z, x, y])

uuid

uuidの生成は標準ライブラリを使用することで簡単にできます。

import uuid

def make_sys_id():
        return str(uuid.uuid4())

# 実行例
>>> make_sys_id()
'ac441afe-fc2d-4ebb-a9cf-18a49c77ec71'

hash

MD5でハッシュ化してハッシュ値を求めます。

import hashlib

serialized = 'hoge'
md5 = hashlib.md5(serialized.encode('utf-8')).hexdigest()

# 実行例
>>> print(md5)
ea703e7aa1efda0064eaa507d9e8ab7e

※パスワードなど秘匿するべき情報の保存には向いていません。より安全なハッシュ関数・ソルトなどを使用する bcrypt などが便利です。

日付

日付処理を行いたい時の例。

import datetime
from dateutil.relativedelta import relativedelta

# 今日の日付
today_tmp = datetime.date.today()
today = today_tmp.strftime('%Y%m%d')

>>> print(today)
20201225

# 明日の日付
tomorrow_tmp = today_tmp + datetime.timedelta(days=1)
# 昨日の日付
yesterday_tmp = today_tmp - datetime.timedelta(days=1)

# 1ヶ月前の明日の日付
one_month_before = tomorrow_tmp - relativedelta(months=1)
one_month_before = one_month_before.strftime('%Y%m%d')

>>> print(one_month_before)
20201125

# 1ヶ月後の昨日の日付
one_month_later = yesterday_tmp + relativedelta(months=1)
one_month_later = one_month_later.strftime('%Y%m%d')

>>> print(one_month_later)
20210123

バッチの戻り値

tryキャッチの中で処理結果に応じて、バッチの戻り値を出力して終了したい場合。
なお、戻り値の出力は他にも方法があります。

import os

# 例
try:
    処理する内容（例：DB登録）
    # 正常終了
    os._exit(0)
except:
    # 異常終了
    os._exit(99)

自作例外クラス

自作例外クラスを作成し、raiseで例外を投げます。
バッチ処理のtryキャッチに欠かせません。

class BatchError(Exception):
    def __init__(self, m):
        self.message = m
    def __str__(self):
        return self.message

# 例
try:
    処理する内容（例：DB接続）
except:
    e = traceback.format_exc()
    logging.error(e)
    logging.error('DBに接続できないため、処理を終了します')
    raise BatchError("DB接続失敗")

デバッグ

デバッグ方法の例。
個人的なおすすめはpysnooperのライブラリです。

pysnooper

import pysnooper

使用方法は簡単です。デバッグしたい関数に@pysnooper.snoop()をデコレートします。
これでバッチを実行すれば変数の中身など詳細に出力されます。

pprint

json形式でスッキリ見たい場合に、pprintは便利です。

from pprint import pprint

日本語エンコード対策

OSの環境などに依存し、エンコード方式がANSIの場合にPython３で日本語の出力が失敗する場合があります。

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

その他

Pyhonを学ぶにあたり、殆どの入門書では最近のPythonの書き方について書いてある方が少ないと思います。Pythonのレベルアップを行うためには、自分で新しい情報をキャッチアップするのが不可欠です。

f文字列

f文字列はPython3.6で追加されました。フォーマット済み文字列リテラルです。
従来のformat()メソッドで実現していた置換が文字列リテラルで記述できます。

>>> word = "WORLD"
>>> f'HELLO {word}' 
'HELLO WORLD'

>>> today = datetime(year=2020, month=5, day=6)
>>> f"{today:%B %d, %Y}"
'May 06, 2020'

アノテーション

Pythonは動的型付け言語ですが、Python3.5以降より型アノテーションを可能とします。
Pythonの性質として同じ型を揃えないとエラーになるため、以下のコードはエラーになります。

>>> def test(word):
...  return 'Hello' + word
... 
>>> test(1)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 2, in test
TypeError: can only concatenate str (not "int") to str

型アノテーションを利用することで、コードの保守性を高めることができます。
以下の場合は実引数nameの型はstrであり、返り値の型はstrであることが期待されます。
ポイントはあくまでアノテーションができるだけで、エラーチェックは行われません。

>>> def greeting(name: str) -> str:
...     return 'Hello' + name
... 
>>> greeting("apple")
'Hello apple'

dataclasses

dataclassesはPython3.7で追加されました。ユーザー定義のクラスに自動的に追加するデコレータや関数を提供します。

従来のinitで初期化しての書き方が不要になります。

>>> class Animal:
...     def __init__(self, type, age, name):
...         self.type = type
...         self.age = age
...         self.name = name
... 
>>> cat = Animal("cat", 0,"チュール" )
>>> print(cat.type, cat.age, cat.name)
cat 0 チュール

dataclassesを使うと同じことが簡単にできます。

>>> from dataclasses import dataclass
>>> @dataclass
... class Animal:
...  type: str
...  age: int
...  name: str
... 
>>> cat = Animal("cat", 0, "チュール")
>>> print(cat)
Animal(type='cat', age=0, name='チュール')

おわりに

これからの時代はコンテナ技術を前提としたシステムになってくるため、バッチ設計の考え方も変わってきます。

しかし、いくらサーバレスなど先端な技術を使用しようが、運用を考慮していない場合は課題や技術的負債しか残りません。

技術はあくまで手段です。重要なのはビジネスとしてサービスを継続するために、適切なバッチ設計を行い、サービスに支障をきたさない作り込みを行うことです。

バッチ設計とPython