import multiprocessing as mp
import pandas as pd
import numpy as np
import time
from functools import partial
import os
class ParallelDataProcessor:
def __init__(self, n_processes=None):
        """
        並列データ処理クラス
        Args:
            n_processes: 使用するプロセス数（Noneの場合はCPUコア数）
        """
self.n_processes = n_processes or mp.cpu_count()
print(f"使用プロセス数: {self.n_processes}")
def process_csv_files_parallel(self, file_paths, processing_func):
        """
        複数のCSVファイルを並列処理
        Args:
            file_paths: 処理対象ファイルパスのリスト
            processing_func: 各ファイルに適用する処理関数
        Returns:
            処理結果のリスト
        """
start_time = time.time()
with mp.Pool(processes=self.n_processes) as pool:
results = pool.map(processing_func, file_paths)
end_time = time.time()
print(f"並列処理完了: {end_time - start_time:.2f}秒")
return results
def process_dataframe_chunks_parallel(self, df, processing_func, chunk_size=10000):
        """
        大きなDataFrameをチャンクに分割して並列処理
        Args:
            df: 処理対象DataFrame
            processing_func: 各チャンクに適用する処理関数
            chunk_size: チャンクサイズ
        Returns:
            処理結果を結合したDataFrame
        """
# DataFrameをチャンクに分割
chunks = [df[i:i+chunk_size] for i in range(0, len(df), chunk_size)]
print(f"データを{len(chunks)}個のチャンクに分割")
start_time = time.time()
with mp.Pool(processes=self.n_processes) as pool:
processed_chunks = pool.map(processing_func, chunks)
# 結果を結合
result_df = pd.concat(processed_chunks, ignore_index=True)
end_time = time.time()
print(f"チャンク並列処理完了: {end_time - start_time:.2f}秒")
return result_df
# 使用例: CSVファイル処理
def process_single_csv(file_path):
    """単一CSVファイルの処理関数"""
try:
df = pd.read_csv(file_path)
# データクリーニング
df = df.dropna()
df = df[df['value'] > 0]  # 正の値のみ
# 統計計算
stats = {
'file': os.path.basename(file_path),
'rows': len(df),
'mean_value': df['value'].mean(),
'std_value': df['value'].std(),
'max_value': df['value'].max(),
'min_value': df['value'].min()
}
return stats
except Exception as e:
return {'file': os.path.basename(file_path), 'error': str(e)}
# 実行例
if __name__ == "__main__":
processor = ParallelDataProcessor(n_processes=4)
# CSVファイルリスト
csv_files = ['data1.csv', 'data2.csv', 'data3.csv', 'data4.csv']
# 並列処理実行
results = processor.process_csv_files_parallel(csv_files, process_single_csv)
# 結果表示
for result in results:
print(result)

import concurrent.futures
import requests
import json
import time
from typing import List, Dict, Any
import threading
class AdvancedParallelProcessor:
def __init__(self, max_workers=None):
        """
        高度な並列処理クラス
        Args:
            max_workers: 最大ワーカー数
        """
self.max_workers = max_workers or min(32, (os.cpu_count() or 1) + 4)
self.session = requests.Session()
self.lock = threading.Lock()
self.results = []
def parallel_api_calls(self, urls: List[str], timeout=30):
        """
        複数のAPI呼び出しを並列実行
        Args:
            urls: API URLのリスト
            timeout: タイムアウト時間
        Returns:
            API応答結果のリスト
        """
def fetch_url(url):
try:
response = self.session.get(url, timeout=timeout)
response.raise_for_status()
return {
'url': url,
'status_code': response.status_code,
'data': response.json() if response.headers.get('content-type', '').startswith('application/json') else response.text,
'response_time': response.elapsed.total_seconds()
}
except Exception as e:
return {
'url': url,
'error': str(e),
'status_code': None
}
start_time = time.time()
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
# 全てのタスクを投入
future_to_url = {executor.submit(fetch_url, url): url for url in urls}
results = []
for future in concurrent.futures.as_completed(future_to_url):
result = future.result()
results.append(result)
# 進捗表示
with self.lock:
print(f"完了: {len(results)}/{len(urls)} - {result['url']}")
end_time = time.time()
print(f"API並列呼び出し完了: {end_time - start_time:.2f}秒")
return results
def parallel_file_processing(self, file_paths: List[str], processing_func):
        """
        ファイル処理の並列実行（プロセスベース）
        Args:
            file_paths: ファイルパスのリスト
            processing_func: 処理関数
        Returns:
            処理結果のリスト
        """
start_time = time.time()
with concurrent.futures.ProcessPoolExecutor(max_workers=self.max_workers) as executor:
# タスクを投入
future_to_file = {executor.submit(processing_func, file_path): file_path for file_path in file_paths}
results = []
for future in concurrent.futures.as_completed(future_to_file):
try:
result = future.result()
results.append(result)
# 進捗表示
file_path = future_to_file[future]
print(f"処理完了: {os.path.basename(file_path)}")
except Exception as e:
file_path = future_to_file[future]
print(f"処理エラー: {os.path.basename(file_path)} - {str(e)}")
results.append({'file': file_path, 'error': str(e)})
end_time = time.time()
print(f"ファイル並列処理完了: {end_time - start_time:.2f}秒")
return results
# 使用例: Web API データ収集
def collect_api_data():
    """複数のAPIからデータを並列収集"""
processor = AdvancedParallelProcessor(max_workers=10)
# API URLリスト
api_urls = [
'https://api.github.com/users/octocat',
'https://api.github.com/users/defunkt',
'https://api.github.com/users/pjhyett',
# ... 他のURL
]
# 並列API呼び出し
results = processor.parallel_api_calls(api_urls)
# 成功・失敗の集計
successful = [r for r in results if 'error' not in r]
failed = [r for r in results if 'error' in r]
print(f"成功: {len(successful)}件, 失敗: {len(failed)}件")
return successful, failed

import numpy as np
import pandas as pd
from multiprocessing import Pool, Manager, Value
import psutil
import gc
class OptimizedDataProcessor:
def __init__(self):
        """最適化されたデータ処理クラス"""
self.cpu_count = psutil.cpu_count(logical=False)  # 物理コア数
self.memory_gb = psutil.virtual_memory().total / (1024**3)
print(f"システム情報:")
print(f"  物理CPUコア数: {self.cpu_count}")
print(f"  メモリ: {self.memory_gb:.1f}GB")
def calculate_optimal_chunk_size(self, data_size_mb, target_memory_usage_gb=2):
        """
        最適なチャンクサイズを計算
        Args:
            data_size_mb: データサイズ（MB）
            target_memory_usage_gb: 目標メモリ使用量（GB）
        Returns:
            最適なチャンクサイズ
        """
available_memory_gb = min(self.memory_gb * 0.7, target_memory_usage_gb)
chunk_size_mb = (available_memory_gb * 1024) / self.cpu_count
chunk_ratio = chunk_size_mb / data_size_mb
chunk_size = max(1000, int(chunk_ratio * 1000000))  # 最小1000行
print(f"最適チャンクサイズ: {chunk_size:,}行")
return chunk_size
def process_large_dataset_optimized(self, df, processing_func, progress_callback=None):
        """
        大規模データセットの最適化処理
        Args:
            df: 処理対象DataFrame
            processing_func: 処理関数
            progress_callback: 進捗コールバック関数
        Returns:
            処理結果DataFrame
        """
# メモリ使用量を考慮したチャンクサイズ計算
data_size_mb = df.memory_usage(deep=True).sum() / (1024**2)
chunk_size = self.calculate_optimal_chunk_size(data_size_mb)
# データをチャンクに分割
chunks = [df.iloc[i:i+chunk_size].copy() for i in range(0, len(df), chunk_size)]
total_chunks = len(chunks)
print(f"データ分割: {total_chunks}個のチャンク")
# 進捗管理用の共有変数
with Manager() as manager:
progress_counter = manager.Value('i', 0)
# 進捗付き処理関数
def process_chunk_with_progress(chunk_data):
chunk, chunk_index = chunk_data
try:
# 実際の処理実行
result = processing_func(chunk)
# 進捗更新
with progress_counter.get_lock():
progress_counter.value += 1
current_progress = progress_counter.value
# 進捗表示
progress_pct = (current_progress / total_chunks) * 100
print(f"進捗: {current_progress}/{total_chunks} ({progress_pct:.1f}%)")
if progress_callback:
progress_callback(current_progress, total_chunks)
# メモリクリーンアップ
del chunk
gc.collect()
return result
except Exception as e:
print(f"チャンク{chunk_index}処理エラー: {str(e)}")
return pd.DataFrame()  # 空のDataFrameを返す
# チャンクにインデックスを付与
indexed_chunks = [(chunk, i) for i, chunk in enumerate(chunks)]
# 並列処理実行
start_time = time.time()
with Pool(processes=self.cpu_count) as pool:
processed_chunks = pool.map(process_chunk_with_progress, indexed_chunks)
end_time = time.time()
# 結果を結合
valid_chunks = [chunk for chunk in processed_chunks if not chunk.empty]
if valid_chunks:
result_df = pd.concat(valid_chunks, ignore_index=True)
print(f"処理完了: {len(result_df):,}行, 処理時間: {end_time - start_time:.2f}秒")
return result_df
else:
print("処理結果が空です")
return pd.DataFrame()
# 使用例: 大規模データの統計処理
def advanced_statistical_processing(df):
    """高度な統計処理関数"""
# 複数の統計指標を計算
result = df.copy()
# 移動平均（複数期間）
for window in [7, 14, 30]:
result[f'ma_{window}'] = df['value'].rolling(window=window).mean()
# 標準化
result['value_normalized'] = (df['value'] - df['value'].mean()) / df['value'].std()
# 異常値検出（IQR法）
Q1 = df['value'].quantile(0.25)
Q3 = df['value'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
result['is_outlier'] = (df['value'] < lower_bound) | (df['value'] > upper_bound)
# カテゴリ別集計
if 'category' in df.columns:
category_stats = df.groupby('category')['value'].agg(['mean', 'std', 'count'])
result = result.merge(category_stats, left_on='category', right_index=True, suffixes=('', '_category'))
return result
# 実行例
if __name__ == "__main__":
# 大規模データセット生成（テスト用）
np.random.seed(42)
large_df = pd.DataFrame({
'value': np.random.normal(100, 15, 1000000),
'category': np.random.choice(['A', 'B', 'C', 'D'], 1000000),
'date': pd.date_range('2020-01-01', periods=1000000, freq='1min')
})
print(f"データサイズ: {len(large_df):,}行")
# 最適化処理実行
processor = OptimizedDataProcessor()
result = processor.process_large_dataset_optimized(
large_df, 
advanced_statistical_processing
)
print(f"処理結果: {len(result):,}行")

import cv2
from PIL import Image
import numpy as np
from concurrent.futures import ProcessPoolExecutor
import os
class ImageProcessingPipeline:
def __init__(self, max_workers=None):
self.max_workers = max_workers or os.cpu_count()
def process_single_image(self, image_path):
        """単一画像の処理パイプライン"""
try:
# 画像読み込み
image = cv2.imread(image_path)
if image is None:
return {'path': image_path, 'error': '画像読み込み失敗'}
# 前処理
resized = cv2.resize(image, (224, 224))
normalized = resized / 255.0
# 特徴抽出
gray = cv2.cvtColor(resized, cv2.COLOR_BGR2GRAY)
edges = cv2.Canny(gray, 50, 150)
edge_density = np.sum(edges > 0) / (edges.shape[0] * edges.shape[1])
# 品質評価
blur_score = cv2.Laplacian(gray, cv2.CV_64F).var()
# 結果保存
output_path = image_path.replace('.jpg', '_processed.jpg')
cv2.imwrite(output_path, resized)
return {
'input_path': image_path,
'output_path': output_path,
'edge_density': edge_density,
'blur_score': blur_score,
'quality': 'high' if blur_score > 100 else 'low'
}
except Exception as e:
return {'path': image_path, 'error': str(e)}
def process_image_batch(self, image_paths):
        """画像バッチ処理"""
with ProcessPoolExecutor(max_workers=self.max_workers) as executor:
results = list(executor.map(self.process_single_image, image_paths))
# 結果集計
successful = [r for r in results if 'error' not in r]
failed = [r for r in results if 'error' in r]
return successful, failed
# 使用例
pipeline = ImageProcessingPipeline(max_workers=8)
image_files = ['image1.jpg', 'image2.jpg', ...]  # 10万枚のリスト
successful, failed = pipeline.process_image_batch(image_files)
print(f"成功: {len(successful)}枚, 失敗: {len(failed)}枚")