# ===== chunksizeを使った読み込み ===== import pandas as pd # 10万行ずつ読み込む chunk_size = 100000 # read_csvにchunksizeを指定すると、イテレータが返る for chunk in pd.read_csv(‘large_file.csv’, chunksize=chunk_size): # chunkはDataFrame（10万行分） print(f”チャンクサイズ: {len(chunk)}行”) print(f”メモリ使用量: {chunk.memory_usage(deep=True).sum() / 1024**2:.2f} MB”) # ここでデータ処理（例：特定の条件でフィルタ） filtered = chunk[chunk[‘price’] > 1000] print(f”フィルタ後: {len(filtered)}行”) print(“—“)

# ===== チャンクを処理して結合 ===== import pandas as pd chunk_size = 100000 result_chunks = [] # 結果を格納するリスト # 各チャンクを処理 for chunk in pd.read_csv(‘large_file.csv’, chunksize=chunk_size): # 条件でフィルタリング（例：価格1000円以上） filtered = chunk[chunk[‘price’] > 1000] # 結果が空でなければリストに追加 if len(filtered) > 0: result_chunks.append(filtered) # すべてのチャンクを結合 result = pd.concat(result_chunks, ignore_index=True) print(f”結果: {len(result)}行”) print(result.head())

# ===== チャンクごとに集計して合計 ===== import pandas as pd chunk_size = 100000 total_sales = 0 # 売上合計 total_count = 0 # 行数カウント processed_chunks = 0 # 処理済みチャンク数 for chunk in pd.read_csv(‘sales.csv’, chunksize=chunk_size): # 各チャンクで集計 chunk_sales = chunk[‘sales’].sum() chunk_count = len(chunk) # 合計に加算 total_sales += chunk_sales total_count += chunk_count processed_chunks += 1 print(f”チャンク {processed_chunks}: 売上 {chunk_sales:,.0f}円, 行数 {chunk_count:,}行”) print(“=” * 50) print(f”総売上: {total_sales:,.0f}円”) print(f”総行数: {total_count:,}行”) print(f”平均売上: {total_sales / total_count:,.2f}円”)

# ===== カテゴリ別集計（チャンク処理） ===== import pandas as pd from collections import defaultdict chunk_size = 100000 # カテゴリごとの売上を格納する辞書 category_sales = defaultdict(float) category_count = defaultdict(int) for chunk in pd.read_csv(‘sales.csv’, chunksize=chunk_size): # チャンク内でgroupby集計 chunk_summary = chunk.groupby(‘category’).agg({ ‘sales’: ‘sum’, ‘order_id’: ‘count’ }) # 辞書に追加 for category, row in chunk_summary.iterrows(): category_sales[category] += row[‘sales’] category_count[category] += row[‘order_id’] # 結果をDataFrameに変換 result = pd.DataFrame({ ‘category’: list(category_sales.keys()), ‘total_sales’: list(category_sales.values()), ‘order_count’: list(category_count.values()) }) # 売上順にソート result = result.sort_values(‘total_sales’, ascending=False) print(result)

# ===== 結果をファイルに追記 ===== import pandas as pd chunk_size = 100000 output_file = ‘filtered_data.csv’ first_chunk = True # ヘッダーを書くかどうかのフラグ for chunk in pd.read_csv(‘huge_file.csv’, chunksize=chunk_size): # 条件でフィルタ（例：2024年1月のデータ） filtered = chunk[ (chunk[‘date’] >= ‘2024-01-01’) & (chunk[‘date’] < '2024-02-01') ] # 結果があればファイルに追記 if len(filtered) > 0: filtered.to_csv( output_file, mode=’a’, # 追記モード（’w’は上書き） header=first_chunk, # 最初だけヘッダーを書く index=False ) first_chunk = False print(f”処理済み: {chunk.index[-1] + 1:,}行”) print(f”完了！結果は {output_file} に保存されました”)

# ===== dtypeを指定して読み込み ===== import pandas as pd # データ型を辞書で定義 dtypes = { ‘user_id’: ‘int32’, # int64 → int32（メモリ半減） ‘age’: ‘int8’, # 0〜120なのでint8で十分 ‘gender’: ‘category’, # ‘M’/’F’のような限られた値 ‘prefecture’: ‘category’, # 47都道府県（繰り返しが多い） ‘price’: ‘float32’, # float64 → float32（メモリ半減） ‘product_name’: ‘str’ # 文字列 } # dtype指定で読み込み df = pd.read_csv(‘data.csv’, dtype=dtypes) # メモリ使用量を確認 print(“=== カラムごとのメモリ使用量 ===”) print(df.memory_usage(deep=True)) print() print(f”合計: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB”)

# ===== 読み込み後の最適化 ===== import pandas as pd # まず普通に読み込み df = pd.read_csv(‘data.csv’) # 最適化前のメモリ使用量 print(“=== 最適化前 ===”) print(f”合計: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB”) print() print(“データ型:”) print(df.dtypes) print() # 各カラムの値の範囲を確認 print(“数値カラムの範囲:”) for col in df.select_dtypes(include=[‘int64’, ‘float64’]).columns: print(f” {col}: {df[col].min()} 〜 {df[col].max()}”) # 最適化 df[‘user_id’] = df[‘user_id’].astype(‘int32’) # 範囲に合わせて変更 df[‘age’] = df[‘age’].astype(‘int8’) # 0〜120なのでint8 df[‘prefecture’] = df[‘prefecture’].astype(‘category’) # 繰り返しが多い df[‘price’] = df[‘price’].astype(‘float32’) # 精度を落としてOK # 最適化後のメモリ使用量 print(“\n=== 最適化後 ===”) print(f”合計: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB”)

# ===== 自動メモリ最適化関数 ===== import pandas as pd import numpy as np def optimize_memory(df, verbose=True): “”” DataFrameのメモリを自動最適化 Parameters: df: 最適化するDataFrame verbose: 詳細を表示するか Returns: 最適化されたDataFrame “”” start_mem = df.memory_usage(deep=True).sum() / 1024**2 for col in df.columns: col_type = df[col].dtype # 数値型の最適化 if col_type != ‘object’ and col_type.name != ‘category’: c_min = df[col].min() c_max = df[col].max() # 整数型 if str(col_type)[:3] == ‘int’: if c_min >= 0: # 非負整数 if c_max < 255: df[col] = df[col].astype(np.uint8) elif c_max < 65535: df[col] = df[col].astype(np.uint16) elif c_max < 4294967295: df[col] = df[col].astype(np.uint32) else: # 負の値あり 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 str(col_type)[:5] == 'float': if c_min > np.finfo(np.float32).min and c_max < np.finfo(np.float32).max: df[col] = df[col].astype(np.float32) # 文字列型 → カテゴリ型への変換 elif col_type == 'object': num_unique = df[col].nunique() num_total = len(df[col]) # ユニーク値が全体の50%以下ならカテゴリ化 if num_unique / num_total < 0.5: df[col] = df[col].astype('category') end_mem = df.memory_usage(deep=True).sum() / 1024**2 if verbose: print(f"最適化前: {start_mem:.2f} MB") print(f"最適化後: {end_mem:.2f} MB") print(f"削減率: {100 * (start_mem - end_mem) / start_mem:.1f}%") return df # 使用例 df = pd.read_csv('data.csv') df = optimize_memory(df)

# ===== category型の効果を確認 ===== import pandas as pd import numpy as np # サンプルデータ作成 n = 1000000 prefectures = [‘東京都’, ‘大阪府’, ‘愛知県’, ‘北海道’, ‘福岡県’] * (n // 5) df = pd.DataFrame({‘prefecture’: prefectures}) # object型のメモリ print(“=== object型 ===”) print(f”メモリ: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB”) # category型に変換 df[‘prefecture’] = df[‘prefecture’].astype(‘category’) # category型のメモリ print(“\n=== category型 ===”) print(f”メモリ: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB”)

# ===== Daskで大容量ファイル読み込み ===== import dask.dataframe as dd # Daskで読み込み（この時点ではまだ読み込まない = 遅延評価） ddf = dd.read_csv(‘large_file.csv’) # DataFrameの情報を確認 print(f”カラム: {ddf.columns.tolist()}”) print(f”パーティション数: {ddf.npartitions}”) # 最初の5行を表示（この時点で必要な部分だけ読み込む） print(ddf.head()) # 集計（遅延評価 = まだ計算しない） total_sales = ddf[‘sales’].sum() print(type(total_sales)) # dask.dataframe.core.Scalar # 実際に計算を実行（.compute()で計算開始） result = total_sales.compute() print(f”売上合計: {result:,.0f}円”)

# ===== 複数CSVを一度に読み込む ===== import dask.dataframe as dd # ワイルドカードで複数ファイルを指定 # data/sales_2024_01.csv, sales_2024_02.csv, … を全部読み込む ddf = dd.read_csv(‘data/sales_*.csv’) # フィルタリング filtered = ddf[ddf[‘price’] > 1000] # グループ集計 result = filtered.groupby(‘category’)[‘sales’].sum() # 計算実行（並列処理される） summary = result.compute() print(summary.sort_values(ascending=False))

# ===== 1行ずつ処理（ストリーミング） ===== import csv # 集計用変数 total_sales = 0 count = 0 # ファイルを1行ずつ読み込み with open(‘sales.csv’, ‘r’, encoding=’utf-8′) as f: reader = csv.DictReader(f) # ヘッダー行を自動的に辞書のキーに for row in reader: # 各行を処理 total_sales += float(row[‘sales’]) count += 1 # 10万行ごとに進捗表示 if count % 100000 == 0: print(f”処理済み: {count:,}行”) print(“=” * 40) print(f”総売上: {total_sales:,.0f}円”) print(f”総行数: {count:,}行”) print(f”平均売上: {total_sales / count:,.2f}円”)

# ===== ストリーミングでフィルタリング ===== import csv input_file = ‘large_file.csv’ output_file = ‘filtered.csv’ # 出力ファイルを準備 with open(output_file, ‘w’, newline=”, encoding=’utf-8′) as out_f: writer = None # 入力ファイルを1行ずつ処理 with open(input_file, ‘r’, encoding=’utf-8′) as in_f: reader = csv.DictReader(in_f) for row in reader: # 条件でフィルタ（例：年齢20以上） if int(row[‘age’]) >= 20: # 最初の行でwriterを初期化（ヘッダー出力） if writer is None: writer = csv.DictWriter(out_f, fieldnames=reader.fieldnames) writer.writeheader() writer.writerow(row) print(f”完了！結果は {output_file} に保存されました”)

# ===== 完全な実装例 ===== import pandas as pd from datetime import datetime from collections import defaultdict # 設定 input_file = ‘sales_10gb.csv’ output_file = ‘summary_2024_01.csv’ chunk_size = 100000 # dtype指定でメモリ削減 dtypes = { ‘order_id’: ‘int32’, ‘product_id’: ‘int32’, ‘product_name’: ‘str’, # 後でcategory化 ‘quantity’: ‘int16’, ‘price’: ‘float32’, ‘sales’: ‘float32’ } # 集計用の辞書 product_sales = defaultdict(float) product_count = defaultdict(int) # 処理開始 total_rows = 0 start_time = datetime.now() print(“処理開始…”) # チャンクごとに処理 for i, chunk in enumerate(pd.read_csv( input_file, chunksize=chunk_size, dtype=dtypes, parse_dates=[‘order_date’] # 日付カラムを自動パース )): # 2024年1月のデータだけフィルタ filtered = chunk[ (chunk[‘order_date’] >= ‘2024-01-01’) & (chunk[‘order_date’] < '2024-02-01') ] # 商品別に集計（groupbyで高速化） if len(filtered) > 0: chunk_summary = filtered.groupby(‘product_name’).agg({ ‘sales’: ‘sum’, ‘order_id’: ‘count’ }) for product, row in chunk_summary.iterrows(): product_sales[product] += row[‘sales’] product_count[product] += row[‘order_id’] # 進捗表示 total_rows += len(chunk) if (i + 1) % 10 == 0: # 10チャンクごと elapsed = (datetime.now() – start_time).total_seconds() speed = total_rows / elapsed print(f”処理済み: {total_rows:,}行 ({elapsed:.1f}秒, {speed:,.0f}行/秒)”) # 結果をDataFrameに変換 result_df = pd.DataFrame({ ‘product_name’: list(product_sales.keys()), ‘total_sales’: list(product_sales.values()), ‘order_count’: list(product_count.values()) }) # 売上順にソート result_df = result_df.sort_values(‘total_sales’, ascending=False) # CSVに保存 result_df.to_csv(output_file, index=False) # 結果表示 print(“\n” + “=” * 50) print(“処理完了！”) print(f”総行数: {total_rows:,}行”) print(f”処理時間: {(datetime.now() – start_time).total_seconds():.1f}秒”) print(f”\n売上トップ10:”) print(result_df.head(10).to_string(index=False))