# ===== エラーハンドリングの比較 ===== import pandas as pd # ❌ エラーハンドリングなし（エラーで止まる） def load_csv_bad(filename): df = pd.read_csv(filename) return df # ✅ エラーハンドリングあり（エラーでも止まらない） def load_csv_good(filename): try: df = pd.read_csv(filename) print(f”✅ ファイル読み込み成功: {filename}”) return df except FileNotFoundError: print(f”❌ ファイルが見つかりません: {filename}”) return None except pd.errors.EmptyDataError: print(f”❌ ファイルが空です: {filename}”) return None except Exception as e: print(f”❌ 予期しないエラー: {e}”) return None # 使用例 df = load_csv_good(‘存在しないファイル.csv’)

# ===== 複数の例外をキャッチ ===== def safe_divide(a, b): “””安全な割り算””” try: result = a / b print(f”✅ 計算成功: {a} ÷ {b} = {result}”) return result except ZeroDivisionError: print(“❌ 0で割ることはできません”) return None except TypeError: print(“❌ 数値以外が入力されました”) return None except Exception as e: print(f”❌ 予期しないエラー: {e}”) return None # テスト safe_divide(10, 2) # 成功 safe_divide(10, 0) # ZeroDivisionError safe_divide(10, “a”) # TypeError

# ===== finallyで必ずクリーンアップ ===== def read_file_with_cleanup(filename): “””ファイルを読んで必ずクローズする””” f = None try: f = open(filename, ‘r’) content = f.read() print(“✅ ファイル読み込み成功”) return content except FileNotFoundError: print(“❌ ファイルが見つかりません”) return None finally: # エラーが出ても出なくても、ファイルをクローズ if f: f.close() print(“🔒 ファイルをクローズしました”) # 使用例 read_file_with_cleanup(‘test.txt’)

# ===== with文を使うと自動でクローズされる ===== def read_file_best(filename): try: with open(filename, ‘r’) as f: content = f.read() print(“✅ ファイル読み込み成功”) return content except FileNotFoundError: print(“❌ ファイルが見つかりません”) return None # ファイルは自動でクローズされる

# ===== カスタム例外クラス ===== class DataValidationError(Exception): “””データ検証エラー””” pass def validate_age(age): “””年齢を検証””” try: age = int(age) if age < 0 or age > 150: raise DataValidationError(f”年齢が範囲外です: {age}”) print(f”✅ 年齢検証OK: {age}”) return age except ValueError: print(f”❌ 年齢が数値ではありません: {age}”) return None except DataValidationError as e: print(f”❌ {e}”) return None # テスト validate_age(25) # OK validate_age(200) # 範囲外 validate_age(“abc”) # 数値ではない

# ===== loggingの基本 ===== import logging # ロガーの設定 logging.basicConfig( level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’ ) # ログ出力 logging.debug(“デバッグ情報”) # 表示されない（INFOレベル以上のみ） logging.info(“処理開始”) logging.warning(“警告メッセージ”) logging.error(“エラーが発生”) logging.critical(“致命的なエラー”)

# ===== ファイルにログを保存 ===== import logging # ファイルにログを保存する設定 logging.basicConfig( level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’, handlers=[ logging.FileHandler(‘etl.log’, encoding=’utf-8′), # ファイルに保存 logging.StreamHandler() # コンソールにも表示 ] ) logging.info(“ETL処理を開始します”) logging.info(“データを読み込み中…”) logging.warning(“欠損値が見つかりました”) logging.info(“ETL処理が完了しました”) # ログファイルの内容を確認 print(“\n=== ログファイル（etl.log）の内容 ===”) with open(‘etl.log’, ‘r’, encoding=’utf-8′) as f: print(f.read())

# ===== ローテーション付きログ ===== import logging from logging.handlers import RotatingFileHandler # ロガーを作成 logger = logging.getLogger(‘etl_pipeline’) logger.setLevel(logging.DEBUG) # フォーマットを定義 formatter = logging.Formatter( ‘%(asctime)s – %(name)s – %(levelname)s – %(funcName)s:%(lineno)d – %(message)s’, datefmt=’%Y-%m-%d %H:%M:%S’ ) # ファイルハンドラ（ローテーション付き） file_handler = RotatingFileHandler( ‘etl_pipeline.log’, maxBytes=10*1024*1024, # 10MB backupCount=5, # 5世代保持 encoding=’utf-8′ ) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(formatter) # コンソールハンドラ console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # コンソールはINFO以上のみ console_handler.setFormatter(formatter) # ロガーにハンドラを追加 logger.addHandler(file_handler) logger.addHandler(console_handler) # 使用例 logger.debug(“デバッグ情報（ファイルのみに記録）”) logger.info(“処理開始”) logger.error(“エラー発生”)

# ===== 基本的なリトライ ===== import time import logging logging.basicConfig(level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’) logger = logging.getLogger(__name__) def retry_function(func, max_retries=3, delay=1): “”” 関数をリトライする Parameters: ———– func : function 実行する関数 max_retries : int 最大リトライ回数 delay : int リトライ間隔（秒） “”” for attempt in range(max_retries): try: result = func() logger.info(f”✅ 成功（{attempt + 1}回目）”) return result except Exception as e: logger.warning(f”❌ 失敗（{attempt + 1}回目）: {e}”) if attempt < max_retries - 1: logger.info(f"⏳ {delay}秒後にリトライします...") time.sleep(delay) else: logger.error(f"❌ {max_retries}回リトライしましたが失敗しました") raise # テスト用の関数（3回目で成功） attempt_count = 0 def unstable_function(): global attempt_count attempt_count += 1 if attempt_count < 3: raise Exception(f"一時的なエラー（{attempt_count}回目）") return "成功！" # 実行 result = retry_function(unstable_function, max_retries=3, delay=1)

# ===== リトライデコレータ ===== import time import logging from functools import wraps logging.basicConfig(level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’) logger = logging.getLogger(__name__) def retry(max_retries=3, delay=1, backoff=2): “”” リトライデコレータ Parameters: ———– max_retries : int 最大リトライ回数 delay : int 初回リトライまでの待機時間（秒） backoff : int リトライごとの待機時間の倍率（指数バックオフ） “”” def decorator(func): @wraps(func) def wrapper(*args, **kwargs): current_delay = delay for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: logger.warning(f”❌ {func.__name__} 失敗（{attempt + 1}回目）: {e}”) if attempt < max_retries - 1: logger.info(f"⏳ {current_delay}秒後にリトライ...") time.sleep(current_delay) current_delay *= backoff # 待機時間を倍に else: logger.error(f"❌ {func.__name__} が{max_retries}回失敗しました") raise return wrapper return decorator # 使用例 @retry(max_retries=3, delay=1, backoff=2) def fetch_data_from_api(): """APIからデータ取得（リトライ付き）""" import random if random.random() < 0.7: # 70%の確率で失敗 raise Exception("API接続エラー") logger.info("✅ APIからデータ取得成功") return {"data": [1, 2, 3]} # 実行 data = fetch_data_from_api()

# ===== tenacityライブラリを使用 ===== from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import logging logging.basicConfig(level=logging.INFO, format=’%(asctime)s – %(levelname)s – %(message)s’) logger = logging.getLogger(__name__) # 特定の例外のみリトライ @retry( stop=stop_after_attempt(3), # 最大3回 wait=wait_exponential(multiplier=1), # 指数バックオフ（1秒、2秒、4秒…） retry=retry_if_exception_type(ConnectionError), # ConnectionErrorのみリトライ reraise=True ) def connect_to_database(): “””データベース接続（リトライ付き）””” logger.info(“データベースに接続中…”) import random if random.random() < 0.5: raise ConnectionError("接続タイムアウト") logger.info("✅ データベース接続成功") return "connected" # 実行 try: result = connect_to_database() except ConnectionError: logger.error("データベース接続に失敗しました")

# ===== 堅牢なETLパイプライン ===== import pandas as pd import logging from logging.handlers import RotatingFileHandler import time from functools import wraps # ===================================== # 1. ロギング設定 # ===================================== def setup_logger(name, log_file, level=logging.INFO): “””ロガーのセットアップ””” logger = logging.getLogger(name) logger.setLevel(level) # フォーマット formatter = logging.Formatter( ‘%(asctime)s – %(levelname)s – %(funcName)s – %(message)s’, datefmt=’%Y-%m-%d %H:%M:%S’ ) # ファイルハンドラ file_handler = RotatingFileHandler( log_file, maxBytes=10*1024*1024, backupCount=5, encoding=’utf-8′ ) file_handler.setFormatter(formatter) # コンソールハンドラ console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger # ロガーを作成 logger = setup_logger(‘etl_pipeline’, ‘etl_pipeline.log’) # ===================================== # 2. リトライデコレータ # ===================================== def retry(max_retries=3, delay=1, backoff=2): “””リトライデコレータ””” def decorator(func): @wraps(func) def wrapper(*args, **kwargs): current_delay = delay for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: logger.warning(f”リトライ {attempt + 1}/{max_retries}: {e}”) if attempt < max_retries - 1: time.sleep(current_delay) current_delay *= backoff else: raise return wrapper return decorator # ===================================== # 3. ETL関数 # ===================================== @retry(max_retries=3, delay=1) def extract_data(file_path): """データ抽出""" logger.info(f"データ抽出開始: {file_path}") try: df = pd.read_csv(file_path) logger.info(f"✅ データ抽出成功: {len(df)}件") return df except FileNotFoundError: logger.error(f"❌ ファイルが見つかりません: {file_path}") raise def transform_data(df): """データ変換""" logger.info("データ変換開始") try: # 欠損値チェック missing_count = df.isnull().sum().sum() if missing_count > 0: logger.warning(f”⚠️ 欠損値が{missing_count}個見つかりました”) df = df.fillna(0) logger.info(“欠損値を0で補完しました”) # 変換処理 if ‘金額’ in df.columns: df[‘金額’] = df[‘金額’] * 1.1 # 税込み価格に変換 logger.info(f”✅ データ変換成功: {len(df)}件”) return df except KeyError as e: logger.error(f”❌ 必要なカラムがありません: {e}”) raise @retry(max_retries=3, delay=2) def load_data(df, output_path): “””データロード””” logger.info(f”データロード開始: {output_path}”) try: df.to_csv(output_path, index=False, encoding=’utf-8-sig’) logger.info(f”✅ データロード成功: {len(df)}件”) return True except PermissionError: logger.error(f”❌ 書き込み権限がありません: {output_path}”) raise # ===================================== # 4. メイン処理 # ===================================== def run_etl_pipeline(input_file, output_file): “””ETLパイプライン実行””” logger.info(“=” * 50) logger.info(“ETLパイプライン開始”) logger.info(“=” * 50) start_time = time.time() try: # Extract df = extract_data(input_file) # Transform df = transform_data(df) # Load load_data(df, output_file) # 実行時間 elapsed_time = time.time() – start_time logger.info(“=” * 50) logger.info(f”✅ ETLパイプライン完了（{elapsed_time:.2f}秒）”) logger.info(“=” * 50) return True except Exception as e: elapsed_time = time.time() – start_time logger.error(“=” * 50) logger.error(f”❌ ETLパイプライン失敗（{elapsed_time:.2f}秒）”) logger.error(f”エラー内容: {e}”) logger.error(“=” * 50) return False # ===================================== # 5. 実行 # ===================================== if __name__ == ‘__main__’: # テストデータを作成 test_df = pd.DataFrame({ ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’], ‘金額’: [100, 150, 80] }) test_df.to_csv(‘input.csv’, index=False) # ETL実行 success = run_etl_pipeline(‘input.csv’, ‘output.csv’)