日: 2026年7月29日

まずはお困りごとに寄り添います。ファイルを大量に処理したいが、依存ライブラリを増やせない。並列化で速くしたいけれど、失敗時の扱いやログ・再実行が心配──そんな実務の壁に対して、Python標準ライブラリだけで安全に運用できるワークフローを示します。小〜中規模の運用を想定し、設定・ログ・検証と組み合わせて現場でそのまま使える手順を紹介します。

この記事の狙いと対象

外部ライブラリに頼らず、pathlib・datetime・concurrent.futures を中心に、ファイル入出力、バッチ推論、前処理を安全に並列化する手順を解説します。対象はAIを仕事に活かしたい実務担当者・個人事業主・中小企業の担当者です。

主要ライブラリの短い説明と使いどころ

ライブラリ	短い説明	現場での使いどころ
pathlib	OS差を吸収する高レベルなファイル操作API	入力ディレクトリ走査、出力パス生成、原子的なファイル移動
datetime	日時処理とフォーマット	ログや出力ファイル名のタイムスタンプ、ローテーション管理
concurrent.futures	スレッド／プロセスプールの統一インターフェース	I/OバウンドはThreadPoolExecutor、CPUバウンドはProcessPoolExecutorで並列化

ワークフローの全体像（ステップ）

1. 設定読み込み（パス・並列数・タイムアウト・再試行回数）
2. 入力ディレクトリのスキャン（pathlib）
3. バッチ分割（ファイルをN件ずつ）
4. 並列実行（concurrent.futures）
5. 出力格納（安全な一時ファイル→移動）
6. 後処理（成功/失敗の整理、再キュー化）

設計判断：Thread vs Process と max_workers の決め方

観点	判断基準	実務での目安
I/O vs CPU	処理が待ち中心（ファイル/ネットワーク）ならI/Oバウンド、演算中心ならCPUバウンド	I/O -> ThreadPoolExecutor、CPU -> ProcessPoolExecutor
max_workers	CPUコア数、メモリ、外部APIのレート制限を考慮	CPU: max(1, cpu_count() – 1)。I/O: 2〜10倍の試行が現実的（監視しながら調整）

堅牢性の組み込み（実装方針）

• タイムアウト: future.result(timeout=…) で個別タスクに制限をかける
• 例外収集: 各futureの例外を集めてログと再実行キューを作る
• 再実行戦略: 固定回数のリトライ + 緩やかなバックオフ
• 部分失敗時の回復: 失敗ファイルを別ディレクトリに移動して再キュー化
• キャンセル: シャットダウン時はfuture.cancel()を呼ぶが、すでに実行中のプロセスは中断されない点に注意

実践：サンプルスクリプト骨子

以下は現場でそのまま貼れる最小限のスクリプト骨子です。実務では設定（YAML/JSON）や詳細なログ設定を追加してください。

#!/usr/bin/env python3
import argparse
import logging
from pathlib import Path
from datetime import datetime
import concurrent.futures
import multiprocessing
import time

# --- 設定 ---
DEFAULT_TIMEOUT = 60  # 秒
DEFAULT_RETRIES = 2

# --- ユーティリティ ---
def timestamp():
    return datetime.now().strftime('%Y%m%d_%H%M%S')

# 単一ファイルの処理（ユーザー実装部分）
def process_file(file_path: Path, out_dir: Path) -> dict:
    """ファイルを読み、何らかの処理をし、出力ファイルを返す。"""
    # 例: I/O中心のダミー処理
    data = file_path.read_bytes()
    # 模擬処理時間
    time.sleep(0.1)
    out_path = out_dir / f"{file_path.stem}_proc_{timestamp()}{file_path.suffix}"
    out_path.write_bytes(data)
    return {"input": str(file_path), "output": str(out_path)}

# ワーカー実行ラッパー（リトライを含む）
def worker_with_retry(file_path: Path, out_dir: Path, timeout: int, retries: int):
    attempt = 0
    last_exc = None
    while attempt <= retries:
        attempt += 1
        try:
            return process_file(file_path, out_dir)
        except Exception as e:
            last_exc = e
            logging.warning('Failed %s attempt=%d error=%s', file_path, attempt, e)
            time.sleep(1 * attempt)  # 簡易バックオフ
    raise last_exc

# メイン実行関数
def run(args):
    in_dir = Path(args.input).expanduser()
    out_dir = Path(args.output).expanduser()
    out_dir.mkdir(parents=True, exist_ok=True)

    files = sorted([p for p in in_dir.iterdir() if p.is_file()])
    if not files:
        logging.info('No files found in %s', in_dir)
        return

    # Executorの選択
    is_cpu_bound = args.cpu_bound
    max_workers = args.max_workers or (max(1, multiprocessing.cpu_count() - 1) if is_cpu_bound else min(32, len(files)))
    executor_cls = concurrent.futures.ProcessPoolExecutor if is_cpu_bound else concurrent.futures.ThreadPoolExecutor

    # 並列実行と堅牢な収集
    futures = []
    failed = []
    with executor_cls(max_workers=max_workers) as ex:
        for f in files:
            fut = ex.submit(worker_with_retry, f, out_dir, args.timeout, args.retries)
            futures.append((f, fut))

        for fpath, fut in futures:
            try:
                res = fut.result(timeout=args.timeout + 5)
                logging.info('Success: %s -> %s', res['input'], res['output'])
            except concurrent.futures.TimeoutError:
                logging.error('Timeout: %s', fpath)
                # ここでキャンセルを試みる
                fut.cancel()
                failed.append((fpath, 'timeout'))
            except Exception as e:
                logging.exception('Failed: %s', fpath)
                failed.append((fpath, str(e)))

    # 失敗ファイルを再キュー化するための出力
    if failed:
        failed_dir = out_dir / 'failed'
        failed_dir.mkdir(exist_ok=True)
        for fpath, reason in failed:
            # 元ファイルを移動して記録（ロールフォワード用）
            target = failed_dir / fpath.name
            try:
                fpath.rename(target)
            except Exception:
                logging.warning('Could not move failed file: %s', fpath)
        logging.info('Failed count: %d', len(failed))

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='並列ファイル処理ワークフロー（標準ライブラリ）')
    parser.add_argument('--input', required=True)
    parser.add_argument('--output', required=True)
    parser.add_argument('--cpu-bound', action='store_true', help='CPUバウンド処理なら指定')
    parser.add_argument('--max-workers', type=int, default=None)
    parser.add_argument('--timeout', type=int, default=DEFAULT_TIMEOUT)
    parser.add_argument('--retries', type=int, default=DEFAULT_RETRIES)
    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
    run(args)

運用連携ポイント

• 設定/ロギング: 第116回で示した共通設定を読み込むことで、運用側の統一ログ形式と連携できます。
• データ検証: 第118回の検証スクリプトを処理の冒頭／末尾に組み込んで、入力・出力のサニティチェックを自動化します。
• run-as-script の設計: CLIでinput/output/cpu-bound/max-workers/timeout/retriesを受け取ると運用が楽になります。
• ワークフロー接続: systemd timer / cron に繋ぐ際は、ログのローテーションと失敗時の通知（メール/Slack）を必ず追加してください。

テストと検証の方針

• ローカルで小規模データ（10〜100ファイル）で動作確認し、並列数を変えて速度とエラー率を観察する。
• モック/フェイクIO: ファイル書き込みや外部API呼び出しをモックして、再現性のある単体テストを作る。
• ユニットテスト: worker関数は副作用を分離してテスト可能にし、失敗パターンを網羅する。

実務上の注意点とチェックリスト

項目	確認・対策
ファイルロック・競合	処理中の一時ディレクトリを用意し、成功時に原子的に移動する
プラットフォーム差	pathlibを使い、改行やパス長に注意。Windowsではパス長に制限がある
外部APIレート制限	並列数を低く抑えるか、リトライ/バックオフを実装する
メモリリーク	大きいファイルはストリーミング処理に変更する。定期的にプロセスを再起動する運用も検討

小さな運用からの拡張提案（次の一歩）

• 軽量ワークフロー化: systemd timer / cron で定期実行、失敗時の通知を組み合わせる
• 必要に応じた拡張: より高度な並列制御には asyncio（I/O特化）、joblib（数値計算）などを評価
• オーケストレーション: 将来的に複雑化したらAirflowやPrefectの導入を検討する（ただし小規模運用では導入コストを衡量）

まとめ

pathlib、datetime、concurrent.futures は標準ライブラリだけで安全かつ実務的なファイル処理ワークフローを構築するのに十分です。重要なのは並列化の前に設計判断（I/OかCPUか、最大ワーカー数、タイムアウトと再試行戦略）を明確にすること。小規模運用ではまず標準ライブラリで実装し、ログ・検証・再試行の仕組みを整えてから段階的に拡張してください。次回はこのスクリプトを systemd timer / cron に接続する手順と運用上の小さな自動化を扱います。