🐍 کتابخانه مرجع پایتون

# نصب (در محیط واقعی)
# pip install dask[complete]

import dask
import dask.dataframe as dd
import dask.array as da
import dask.bag as db
import pandas as pd
import numpy as np

print(f"Dask version: {dask.__version__}")
print("Dask imported successfully!")

import pandas as pd
import dask.dataframe as dd

# ===== Pandas: بارگذاری کامل در حافظه =====
# df = pd.read_csv('large_file.csv')  # همه داده در RAM
# result = df.groupby('column').sum()

# ===== Dask: بارگذاری تکه‌تکه (lazy) =====
# ddf = dd.read_csv('large_file.csv')  # فقط برنامه‌ریزی
# result = ddf.groupby('column').sum().compute()  # اجرا با compute()

# ===== تفاوت‌های کلیدی =====
print("Pandas:")
print("  - داده را کامل در RAM بارگذاری می‌کند")
print("  - عملیات به صورت فوری (eager) اجرا می‌شوند")
print("  - برای داده‌های کوچک و متوسط مناسب است")

print("\nDask:")
print("  - داده را به صورت تکه‌تکه (chunk) پردازش می‌کند")
print("  - عملیات به صورت تنبل (lazy) و با compute() اجرا می‌شوند")
print("  - برای داده‌های بزرگ‌تر از RAM و محاسبات موازی مناسب است")

import dask.dataframe as dd
import pandas as pd
import numpy as np

# ===== ایجاد داده نمونه =====
# در عمل از dd.read_csv() برای فایل‌های بزرگ استفاده می‌شود
# ddf = dd.read_csv('large_file.csv', blocksize='100MB')

# ===== ایجاد Dask DataFrame از Pandas =====
pdf = pd.DataFrame({
    'id': range(1, 101),
    'name': ['User_' + str(i) for i in range(1, 101)],
    'value': np.random.randint(0, 100, 100)
})
ddf = dd.from_pandas(pdf, npartitions=4)

print("Dask DataFrame:")
print(ddf)
print(f"Number of partitions: {ddf.npartitions}")

# ===== عملیات روی Dask DataFrame (lazy) =====
filtered = ddf[ddf['value'] > 50]
grouped = filtered.groupby('name')['value'].mean()

print("\nOperations are lazy (no computation yet):")
print("  - filter: lazy")
print("  - groupby: lazy")

# ===== اجرا با compute() =====
result = grouped.compute()
print("\nAfter compute():")
print(result.head())

# ===== استفاده از head() برای مشاهده سریع =====
print("\nFirst 5 rows (head):")
print(ddf.head())

import dask.array as da
import numpy as np

# ===== ایجاد Dask Array =====
# یک آرایه 10000x10000 با تکه‌های 1000x1000
x = da.random.random((10000, 10000), chunks=(1000, 1000))
print("Dask Array:")
print(x)
print(f"Chunks: {x.chunks}")

# ===== عملیات روی Dask Array (lazy) =====
mean = x.mean()
std = x.std()
sum_all = x.sum()

print("\nOperations are lazy:")
print(f"  mean: {mean}")
print(f"  std: {std}")
print(f"  sum: {sum_all}")

# ===== اجرا با compute() =====
print("\nAfter compute():")
print(f"  Mean: {mean.compute():.4f}")
print(f"  Std: {std.compute():.4f}")
print(f"  Sum: {sum_all.compute():.2f}")

# ===== عملیات ماتریسی =====
y = da.random.random((10000, 10000), chunks=(1000, 1000))
result = da.matmul(x, y)
print(f"\nMatrix multiplication shape: {result.shape}")

import dask
from dask.delayed import delayed
import time

# ===== توابع نمونه =====
def add_two(x):
    time.sleep(0.1)  # شبیه‌سازی کار سنگین
    return x + 2

def multiply_four(x):
    time.sleep(0.1)
    return x * 4

def sum_numbers(numbers):
    return sum(numbers)

# ===== استفاده از delayed =====
data = [1, 5, 8, 10]

# ایجاد گره‌های تاخیری (lazy)
step1 = [delayed(add_two)(i) for i in data]
step2 = [delayed(multiply_four)(j) for j in step1]
total = delayed(sum_numbers)(step2)

print("Lazy computation graph created:")

# ===== اجرا =====
start = time.time()
result = total.compute()
end = time.time()
print(f"Result: {result}")
print(f"Time: {end - start:.2f} seconds")

# ===== محاسبه موازی با چندین تابع =====
@delayed
def process_file(filename):
    # شبیه‌سازی پردازش فایل
    time.sleep(0.2)
    return f"Processed {filename}"

files = ['file1.csv', 'file2.csv', 'file3.csv', 'file4.csv']
results = [process_file(f) for f in files]
combined = delayed(sum)(results)

print(f"\nParallel file processing: {combined.compute()}")

import dask.bag as db
import json

# ===== ایجاد داده نمونه =====
data = [
    {'id': 1, 'name': 'Saba', 'score': 95},
    {'id': 2, 'name': 'Ali', 'score': 88},
    {'id': 3, 'name': 'Reza', 'score': 72},
    {'id': 4, 'name': 'Mina', 'score': 91},
    {'id': 5, 'name': 'Hossein', 'score': 65}
]

# ===== ایجاد Bag =====
bag = db.from_sequence(data)
print("Dask Bag:")
print(bag)

# ===== عملیات روی Bag (مانند map/filter) =====
# فیلتر کردن دانشجویان با نمره بالای 80
high_scorers = bag.filter(lambda x: x['score'] > 80)

# استخراج نام‌ها
names = high_scorers.map(lambda x: x['name'])

# ===== اجرا =====
print("\nStudents with score > 80:")
print(list(names.compute()))

# ===== جمع‌آوری آمار =====
total_score = bag.map(lambda x: x['score']).sum()
avg_score = bag.map(lambda x: x['score']).mean()

print(f"\nTotal score: {total_score.compute()}")
print(f"Average score: {avg_score.compute():.2f}")

# ===== پردازش فایل‌های JSON =====
# در عمل: bag = db.read_text('logs/*.json').map(json.loads)
print("\nBag is ideal for processing JSON logs and semi-structured data.")

from dask.distributed import Client
import dask.dataframe as dd
import pandas as pd
import numpy as np

# ===== راه‌اندازی Client محلی =====
# client = Client()  # استفاده از همه هسته‌های CPU
# print(f"Dashboard URL: {client.dashboard_link}")

# ===== تنظیم تعداد Workers =====
# client = Client(n_workers=4, threads_per_worker=2)

# ===== مثال با Client =====
# ایجاد داده بزرگ
pdf = pd.DataFrame({
    'id': range(1, 10001),
    'value': np.random.randint(0, 100, 10000)
})
ddf = dd.from_pandas(pdf, npartitions=8)

# ===== محاسبه با Client =====
# با Client فعال، compute() به صورت توزیع‌شده اجرا می‌شود
result = ddf.groupby('id')['value'].mean().compute()
print("Distributed computation completed!")

print("\nClient features:")
print("  - Dashboard for monitoring")
print("  - Parallel execution across workers")
print("  - Fault tolerance")
print("  - Memory management")

# ===== بستن Client =====
# client.close()

# pip install dask-ml

import dask.dataframe as dd
from dask_ml.preprocessing import StandardScaler
from dask_ml.model_selection import train_test_split
from dask_ml.linear_model import LogisticRegression

# ===== بارگذاری داده بزرگ =====
# ddf = dd.read_csv('large_dataset.csv')

# ===== پیش‌پردازش =====
# X = ddf[['feature1', 'feature2', 'feature3']]
# y = ddf['target']

# ===== تقسیم داده =====
# X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# ===== مقیاس‌سازی =====
# scaler = StandardScaler()
# X_train_scaled = scaler.fit_transform(X_train)
# X_test_scaled = scaler.transform(X_test)

# ===== آموزش مدل =====
# model = LogisticRegression()
# model.fit(X_train_scaled, y_train)

# ===== پیش‌بینی =====
# predictions = model.predict(X_test_scaled)

print("Dask-ML features:")
print("  - Scikit-learn compatible API")
print("  - Distributed training")
print("  - Scalable preprocessing")
print("  - Out-of-core learning")

print("\nSupported estimators:")
print("  - Linear models")
print("  - Tree-based models")
print("  - Clustering")
print("  - Preprocessing")

print("=" * 50)
print("WHEN TO USE DASK")
print("=" * 50)

print("\n✅ USE DASK WHEN:")
print("  - Data is larger than RAM (> 10GB)")
print("  - You have multiple CPU cores available")
print("  - You need to process many files in parallel")
print("  - Your computations are embarrassingly parallel")
print("  - You're using pandas but need to scale")

print("\n❌ DO NOT USE DASK WHEN:")
print("  - Data fits comfortably in RAM (< 1GB)")
print("  - You need low-latency queries (use database)")
print("  - Your operations are highly sequential")
print("  - You have many small tasks (overhead)")
print("  - You need interactive EDA on small data")

print("\n💡 BEST PRACTICES:")
print("  - Use blocksize for reading large files")
print("  - Call compute() only when you need results")
print("  - Use persist() for data used multiple times")
print("  - Monitor dashboard for performance")
print("  - Start with local cluster, then scale")
print("  - Use appropriate API (DataFrame/Array/Bag)")

print("\n📊 COMPARISON:")
print("  Pandas:  Data fits in RAM, fast for small data")
print("  Dask:    Data larger than RAM, multi-core")
print("  Spark:   Very large clusters, SQL-heavy")
print("  DuckDB:  Fast SQL on files without cluster")

import dask.dataframe as dd
import dask.array as da
import pandas as pd
import numpy as np
import time

class BigDataAnalyzer:
    """کلاس تحلیل داده بزرگ با Dask"""
    
    def __init__(self, n_rows=100000):
        self.n_rows = n_rows
        self.ddf = None
        self.darr = None
        self.results = {}
    
    def create_sample_data(self):
        """ایجاد داده نمونه بزرگ"""
        print(f"Creating {self.n_rows:,} rows of sample data...")
        
        # داده به صورت تکه‌تکه
        pdf = pd.DataFrame({
            'id': range(self.n_rows),
            'value': np.random.randn(self.n_rows),
            'category': np.random.choice(['A', 'B', 'C', 'D'], self.n_rows),
            'group': np.random.choice(['X', 'Y', 'Z'], self.n_rows)
        })
        
        self.ddf = dd.from_pandas(pdf, npartitions=8)
        self.darr = da.from_array(pdf['value'].values, chunks=(10000,))
        return self
    
    def analyze_dataframe(self):
        """تحلیل با Dask DataFrame"""
        print("\n📊 DataFrame Analysis:")
        
        # آمار پایه
        self.results['mean'] = self.ddf['value'].mean().compute()
        self.results['std'] = self.ddf['value'].std().compute()
        print(f"  Mean: {self.results['mean']:.4f}")
        print(f"  Std: {self.results['std']:.4f}")
        
        # گروه‌بندی
        grouped = self.ddf.groupby('category')['value'].mean().compute()
        print("  Mean by category:")
        for cat, val in grouped.items():
            print(f"    {cat}: {val:.4f}")
        
        return self
    
    def analyze_array(self):
        """تحلیل با Dask Array"""
        print("\n🔢 Array Analysis:")
        
        # محاسبات آرایه
        arr_mean = self.darr.mean().compute()
        arr_std = self.darr.std().compute()
        arr_sum = self.darr.sum().compute()
        
        print(f"  Mean: {arr_mean:.4f}")
        print(f"  Std: {arr_std:.4f}")
        print(f"  Sum: {arr_sum:.2f}")
        
        return self
    
    def process_in_chunks(self):
        """پردازش تکه‌تکه (Out-of-Core)"""
        print("\n📦 Chunk Processing:")
        
        # پردازش تکه‌تکه با map_partitions
        def process_chunk(df):
            df['value_squared'] = df['value'] ** 2
            return df
        
        processed = self.ddf.map_partitions(process_chunk)
        result = processed['value_squared'].mean().compute()
        print(f"  Mean of squared values: {result:.4f}")
        
        return self
    
    def filter_and_aggregate(self):
        """فیلتر و تجمیع"""
        print("\n🔍 Filter and Aggregate:")
        
        # فیلتر و محاسبه
        filtered = self.ddf[self.ddf['category'].isin(['A', 'B'])]
        result = filtered.groupby('group')['value'].agg(['mean', 'count']).compute()
        
        print("  Group statistics (categories A & B):")
        for idx, row in result.iterrows():
            print(f"    {idx}: mean={row['mean']:.4f}, count={row['count']}")
        
        return self
    
    def show_summary(self):
        """نمایش خلاصه"""
        print("\n" + "=" * 50)
        print("ANALYSIS SUMMARY")
        print("=" * 50)
        for key, value in self.results.items():
            if isinstance(value, float):
                print(f"  {key}: {value:.4f}")
            else:
                print(f"  {key}: {value}")

# ===== استفاده =====
if __name__ == "__main__":
    start = time.time()
    
    analyzer = BigDataAnalyzer(n_rows=100000)
    analyzer.create_sample_data()
    analyzer.analyze_dataframe()
    analyzer.analyze_array()
    analyzer.process_in_chunks()
    analyzer.filter_and_aggregate()
    analyzer.show_summary()
    
    end = time.time()
    print(f"\n⏱️ Total time: {end - start:.2f} seconds")

print("\n🚀 Dask makes big data processing easy and fast!")
print("   With Dask, you can scale from laptops to clusters.")