Dask

Dask is an open-source Python library for parallel computing that scales your Python analytical workflows from single machines to large clusters. It provides familiar interfaces (like NumPy arrays and Pandas DataFrames) but designed to operate on data that is too large to fit into RAM, or to speed up computations on smaller datasets by utilizing multiple CPU cores.

Key aspects of Dask:

1. Dynamic Task Scheduling: At its core, Dask builds computational graphs of all operations you define. It breaks down large problems into many smaller tasks, organizes them into a directed acyclic graph (DAG), and then executes these tasks in parallel, either locally on a multi-core machine or distributed across a cluster.

2. Big Data Collections: Dask offers high-level collections that mimic existing Python APIs, allowing users to scale up without rewriting their code:
- Dask Array: A chunked, N-dimensional array that extends NumPy. It's suitable for numerical computations on arrays larger than memory.
- Dask DataFrame: A collection of Pandas DataFrames partitioned along an index. It's ideal for processing large tabular datasets, supporting most of the Pandas API.
- Dask Bag: A list-like collection for semi-structured or unstructured data, useful for parallelizing custom computations.

3. Scalability: Dask can seamlessly scale from a single machine using threads or processes to a distributed cluster (e.g., cloud platforms, HPC systems, Kubernetes). This flexibility allows users to choose the appropriate computational resources based on their data size and complexity.

4. Integration: Dask integrates well with the existing Python ecosystem, including NumPy, Pandas, Scikit-learn, and more. This means you can often use your current data science tools and libraries with Dask, just on larger datasets.

5. Lazy Evaluation: Dask operations are 'lazy'. When you perform an operation (e.g., `df.groupby().mean()`), Dask doesn't immediately compute the result. Instead, it builds the task graph. The actual computation only happens when you explicitly call a method like `.compute()`, allowing for optimization before execution.

Benefits of Dask:
- Scalability: Handles datasets larger than available RAM and scales computations across multiple cores or machines.
- Familiar APIs: Leverages existing NumPy and Pandas syntax, making it easy for users to transition.
- Performance: Accelerates computations through parallel processing.
- Flexibility: Works in various environments, from local machines to large distributed clusters.
- Fault Tolerance: Can recover from failures in a distributed setup.

import dask.dataframe as dd
import pandas as pd
import os

 --- 1. Create dummy CSV files for demonstration ---
 In a real scenario, you'd have existing large CSV files
output_dir = "dask_data"
os.makedirs(output_dir, exist_ok=True)

data1 = {'id': [1, 2, 3], 'value': [10, 20, 30], 'category': ['A', 'B', 'A']}
df1 = pd.DataFrame(data1)
df1.to_csv(os.path.join(output_dir, 'part1.csv'), index=False)

data2 = {'id': [4, 5, 6], 'value': [40, 50, 60], 'category': ['B', 'A', 'B']}
df2 = pd.DataFrame(data2)
df2.to_csv(os.path.join(output_dir, 'part2.csv'), index=False)

data3 = {'id': [7, 8, 9], 'value': [70, 80, 90], 'category': ['A', 'B', 'A']}
df3 = pd.DataFrame(data3)
df3.to_csv(os.path.join(output_dir, 'part3.csv'), index=False)

print(f"Created dummy CSV files in: {output_dir}\n")

 --- 2. Use Dask DataFrame to read multiple CSV files ---
 Dask intelligently reads multiple files as partitions of a single DataFrame
ddf = dd.read_csv(os.path.join(output_dir, '-.csv'))

print("Dask DataFrame head (lazy):")
print(ddf.head())
print(f"\nNumber of partitions: {ddf.npartitions}")
print(f"Column dtypes: {ddf.dtypes}")

 --- 3. Perform a computation (e.g., groupby and mean) ---
 This operation is 'lazy' - it builds the task graph but doesn't compute yet
mean_values_by_category = ddf.groupby('category')['value'].mean()

print("\nType of mean_values_by_category (still Dask Series):")
print(type(mean_values_by_category))
print(mean_values_by_category)

 --- 4. Trigger the actual computation using .compute() ---
 Dask will execute the task graph, reading files and performing groupby/mean in parallel
result = mean_values_by_category.compute()

print("\nResult after .compute():")
print(result)
print(f"Type of result (now Pandas Series): {type(result)}")

 --- 5. Clean up dummy files and directory (optional) ---
import shutil
shutil.rmtree(output_dir)
print(f"\nCleaned up directory: {output_dir}")