# ===== チャンク処理の比較 ===== import pandas as pd import numpy as np import time # 大量データを作成（100万件） print(“テストデータを作成中…”) large_df = pd.DataFrame({ ‘ID’: range(1, 1000001), ‘名前’: [f’ユーザー{i}’ for i in range(1, 1000001)], ‘年齢’: np.random.randint(20, 70, 1000000), ‘購入金額’: np.random.randint(1000, 100000, 1000000) }) large_df.to_csv(‘large_data.csv’, index=False) print(f”✅ {len(large_df):,}件のデータを作成しました\n”) # ❌ 一度に全部読み込む（メモリを大量に使う） print(“【方法1】一度に全部読み込む”) start = time.time() df_all = pd.read_csv(‘large_data.csv’) print(f”読み込み時間: {time.time() – start:.2f}秒”) print(f”メモリ使用量: {df_all.memory_usage(deep=True).sum() / 1024**2:.2f} MB\n”) # ✅ チャンクで読み込む（メモリ効率的） print(“【方法2】チャンクで読み込む（10万件ずつ）”) start = time.time() total_sum = 0 chunk_count = 0 for chunk in pd.read_csv(‘large_data.csv’, chunksize=100000): # チャンクごとに処理 total_sum += chunk[‘購入金額’].sum() chunk_count += 1 print(f” チャンク{chunk_count}: {len(chunk):,}件処理完了”) print(f”\n読み込み時間: {time.time() – start:.2f}秒”) print(f”合計購入金額: ¥{total_sum:,}”)

# ===== チャンクで変換して保存 ===== def process_large_file_in_chunks(input_file, output_file, chunksize=100000): “”” 大きなCSVファイルをチャンクで処理して保存 Parameters: ———– input_file : str 入力ファイル output_file : str 出力ファイル chunksize : int チャンクサイズ “”” first_chunk = True for i, chunk in enumerate(pd.read_csv(input_file, chunksize=chunksize)): # データ変換 chunk[‘購入金額（税込）’] = (chunk[‘購入金額’] * 1.1).astype(int) chunk[‘年齢層’] = pd.cut( chunk[‘年齢’], bins=[0, 30, 50, 100], labels=[‘若年層’, ‘中年層’, ‘シニア層’] ) # 保存（最初のチャンクはヘッダー付き、以降は追記） mode = ‘w’ if first_chunk else ‘a’ header = first_chunk chunk.to_csv(output_file, mode=mode, header=header, index=False) first_chunk = False print(f”✅ チャンク{i+1}: {len(chunk):,}件処理完了”) print(f”\n全チャンクの処理が完了しました”) # 実行 process_large_file_in_chunks(‘large_data.csv’, ‘processed_data.csv’, chunksize=100000)

# ===== 基本的な並列処理 ===== from multiprocessing import Pool import time import numpy as np def process_chunk(data): “””チャンクを処理する関数””” # 重い処理をシミュレート result = np.sum(data ** 2) return result # テストデータ data = np.random.randint(1, 100, 10000000) chunks = np.array_split(data, 8) # 8つに分割 # ❌ 逐次処理（1つずつ） print(“【逐次処理】”) start = time.time() results_sequential = [] for chunk in chunks: result = process_chunk(chunk) results_sequential.append(result) time_sequential = time.time() – start print(f”処理時間: {time_sequential:.2f}秒\n”) # ✅ 並列処理（同時に） print(“【並列処理】（4プロセス）”) start = time.time() with Pool(processes=4) as pool: results_parallel = pool.map(process_chunk, chunks) time_parallel = time.time() – start print(f”処理時間: {time_parallel:.2f}秒”) print(f”速度向上: {time_sequential/time_parallel:.1f}倍”)

# ===== DataFrameを並列処理 ===== import pandas as pd from multiprocessing import Pool import numpy as np import time def process_dataframe_chunk(chunk): “””DataFrameのチャンクを処理””” # データ変換 chunk[‘購入金額（税込）’] = chunk[‘購入金額’] * 1.1 chunk[‘ポイント’] = (chunk[‘購入金額’] * 0.01).astype(int) # 年齢層の分類 chunk[‘年齢層’] = pd.cut( chunk[‘年齢’], bins=[0, 30, 50, 100], labels=[‘若年層’, ‘中年層’, ‘シニア層’] ) return chunk def parallel_process_dataframe(df, n_processes=4): “””DataFrameを並列処理””” # DataFrameを分割 chunks = np.array_split(df, n_processes) # 並列処理 with Pool(processes=n_processes) as pool: processed_chunks = pool.map(process_dataframe_chunk, chunks) # 結合 result = pd.concat(processed_chunks, ignore_index=True) return result # テストデータ df = pd.DataFrame({ ‘ID’: range(1, 100001), ‘年齢’: np.random.randint(20, 70, 100000), ‘購入金額’: np.random.randint(1000, 100000, 100000) }) print(“【逐次処理】”) start = time.time() df_sequential = process_dataframe_chunk(df.copy()) time_sequential = time.time() – start print(f”処理時間: {time_sequential:.2f}秒\n”) print(“【並列処理】（4プロセス）”) start = time.time() df_parallel = parallel_process_dataframe(df.copy(), n_processes=4) time_parallel = time.time() – start print(f”処理時間: {time_parallel:.2f}秒”) print(f”速度向上: {time_sequential/time_parallel:.1f}倍”)

# ===== データ型の最適化 ===== import pandas as pd import numpy as np # サンプルデータ df = pd.DataFrame({ ‘ID’: range(1, 1000001), ‘年齢’: np.random.randint(0, 100, 1000000), ‘購入回数’: np.random.randint(0, 50, 1000000), ‘購入金額’: np.random.randint(0, 1000000, 1000000) }) print(“【最適化前】”) print(f”メモリ使用量: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB\n”) print(df.dtypes) print() # データ型を最適化 df_optimized = df.copy() df_optimized[‘ID’] = df_optimized[‘ID’].astype(‘int32’) df_optimized[‘年齢’] = df_optimized[‘年齢’].astype(‘int8’) df_optimized[‘購入回数’] = df_optimized[‘購入回数’].astype(‘int8’) df_optimized[‘購入金額’] = df_optimized[‘購入金額’].astype(‘int32’) print(“【最適化後】”) print(f”メモリ使用量: {df_optimized.memory_usage(deep=True).sum() / 1024**2:.2f} MB\n”) print(df_optimized.dtypes) # 削減率 original = df.memory_usage(deep=True).sum() optimized = df_optimized.memory_usage(deep=True).sum() print(f”\nメモリ削減率: {(1 – optimized / original) * 100:.1f}%”)

# ===== カテゴリ型で大幅削減 ===== import pandas as pd # 都道府県のような重複の多いデータ df = pd.DataFrame({ ‘都道府県’: [‘東京都’] * 250000 + [‘大阪府’] * 250000 + [‘愛知県’] * 250000 + [‘福岡県’] * 250000 }) print(“【通常のobject型】”) memory_before = df.memory_usage(deep=True).sum() / 1024**2 print(f”メモリ使用量: {memory_before:.2f} MB\n”) # カテゴリ型に変換 df[‘都道府県’] = df[‘都道府県’].astype(‘category’) print(“【category型】”) memory_after = df.memory_usage(deep=True).sum() / 1024**2 print(f”メモリ使用量: {memory_after:.2f} MB\n”) print(f”メモリ削減率: {(1 – memory_after / memory_before) * 100:.1f}%”)

# ===== 自動最適化関数 ===== import pandas as pd import numpy as np def optimize_dataframe(df): “”” DataFrameのデータ型を自動最適化 Parameters: ———– df : DataFrame 最適化するDataFrame Returns: ——– DataFrame : 最適化されたDataFrame “”” memory_before = df.memory_usage(deep=True).sum() / 1024**2 print(f”最適化前: {memory_before:.2f} MB”) for col in df.columns: col_type = df[col].dtype # 整数型の最適化 if col_type == ‘int64’: c_min = df[col].min() c_max = df[col].max() if c_min >= np.iinfo(np.int8).min and c_max <= np.iinfo(np.int8).max: df[col] = df[col].astype(np.int8) elif c_min >= np.iinfo(np.int16).min and c_max <= np.iinfo(np.int16).max: df[col] = df[col].astype(np.int16) elif c_min >= np.iinfo(np.int32).min and c_max <= np.iinfo(np.int32).max: df[col] = df[col].astype(np.int32) # 浮動小数点型の最適化 elif col_type == 'float64': df[col] = df[col].astype(np.float32) # オブジェクト型をカテゴリ型に elif col_type == 'object': num_unique = df[col].nunique() num_total = len(df[col]) if num_unique / num_total < 0.5: # ユニーク値が50%未満 df[col] = df[col].astype('category') memory_after = df.memory_usage(deep=True).sum() / 1024**2 print(f"最適化後: {memory_after:.2f} MB") print(f"削減率: {(1 - memory_after / memory_before) * 100:.1f}%") return df # 使用例 df = pd.DataFrame({ 'ID': range(1, 100001), '年齢': np.random.randint(0, 100, 100000), '都道府県': np.random.choice(['東京', '大阪', '愛知'], 100000) }) df = optimize_dataframe(df)

# ===== 必要な列だけ取得 ===== import pandas as pd from sqlalchemy import create_engine engine = create_engine(‘sqlite:///sample.db’) # ❌ 全列を取得（遅い） query_bad = “SELECT * FROM sales” # ✅ 必要な列だけ取得（速い） query_good = “SELECT date, product_id, amount FROM sales” df = pd.read_sql(query_good, engine) print(“必要な列だけを取得しました”)

# ===== SQLで絞り込み ===== # ❌ 全データを取得してからPythonで絞り込み（遅い） query_bad = “SELECT * FROM sales” df = pd.read_sql(query_bad, engine) df = df[df[‘date’] >= ‘2024-01-01’] # ✅ SQLで絞り込んでから取得（速い） query_good = “”” SELECT * FROM sales WHERE date >= ‘2024-01-01’ “”” df = pd.read_sql(query_good, engine) print(“SQLで絞り込んでから取得しました”)

# ===== SQLで集計 ===== # ❌ 全データを取得してからPythonで集計（遅い） query_bad = “SELECT * FROM sales” df = pd.read_sql(query_bad, engine) result = df.groupby(‘product_id’)[‘amount’].sum() # ✅ SQLで集計してから取得（速い） query_good = “”” SELECT product_id, SUM(amount) as total_amount FROM sales GROUP BY product_id “”” df = pd.read_sql(query_good, engine) print(“SQLで集計してから取得しました”)

# ===== 100万件データの高速処理 ===== import pandas as pd import numpy as np import time from multiprocessing import Pool # ===================================== # 1. データ準備 # ===================================== print(“テストデータを作成中…”) df = pd.DataFrame({ ‘ID’: range(1, 1000001), ‘名前’: [f’ユーザー{i}’ for i in range(1, 1000001)], ‘年齢’: np.random.randint(20, 70, 1000000), ‘都道府県’: np.random.choice([‘東京’, ‘大阪’, ‘愛知’, ‘福岡’], 1000000), ‘購入金額’: np.random.randint(1000, 100000, 1000000) }) df.to_csv(‘test_1m.csv’, index=False) print(f”✅ {len(df):,}件のテストデータを作成\n”) # ===================================== # 2. 最適化なし（ベースライン） # ===================================== print(“=” * 50) print(“【最適化なし】”) print(“=” * 50) start = time.time() df = pd.read_csv(‘test_1m.csv’) df[‘購入金額（税込）’] = df[‘購入金額’] * 1.1 df[‘年齢層’] = pd.cut(df[‘年齢’], bins=[0, 30, 50, 100], labels=[‘若年層’, ‘中年層’, ‘シニア層’]) summary = df.groupby([‘都道府県’, ‘年齢層’])[‘購入金額（税込）’].sum().reset_index() summary.to_csv(‘result_no_opt.csv’, index=False) time_no_opt = time.time() – start memory_no_opt = df.memory_usage(deep=True).sum() / 1024**2 print(f”処理時間: {time_no_opt:.2f}秒”) print(f”メモリ使用量: {memory_no_opt:.2f} MB\n”) # ===================================== # 3. チャンク処理 + データ型最適化 # ===================================== print(“=” * 50) print(“【チャンク処理 + データ型最適化】”) print(“=” * 50) start = time.time() dtype_dict = { ‘ID’: ‘int32’, ‘名前’: ‘object’, ‘年齢’: ‘int8’, ‘都道府県’: ‘category’, ‘購入金額’: ‘int32’ } summaries = [] for chunk in pd.read_csv(‘test_1m.csv’, chunksize=100000, dtype=dtype_dict): chunk[‘購入金額（税込）’] = chunk[‘購入金額’] * 1.1 chunk[‘年齢層’] = pd.cut(chunk[‘年齢’], bins=[0, 30, 50, 100], labels=[‘若年層’, ‘中年層’, ‘シニア層’]) chunk_summary = chunk.groupby([‘都道府県’, ‘年齢層’])[‘購入金額（税込）’].sum().reset_index() summaries.append(chunk_summary) summary = pd.concat(summaries, ignore_index=True) summary = summary.groupby([‘都道府県’, ‘年齢層’])[‘購入金額（税込）’].sum().reset_index() summary.to_csv(‘result_optimized.csv’, index=False) time_optimized = time.time() – start print(f”処理時間: {time_optimized:.2f}秒”) print(f”速度改善: {time_no_opt/time_optimized:.1f}倍\n”) print(“=” * 50) print(“【結果サマリー】”) print(“=” * 50) print(f”最適化なし: {time_no_opt:.2f}秒, {memory_no_opt:.2f} MB”) print(f”最適化あり: {time_optimized:.2f}秒, メモリ大幅削減”)