Dask

{
  "description": "\"Parallel/distributed computing. Scale pandas/NumPy beyond memory, parallel DataFrames/Arrays, multi-file processing, task graphs, for larger-than-RAM datasets and parallel workflows.\"",
  "references": {
    "files": [
      "references/arrays.md",
      "references/bags.md",
      "references/best-practices.md",
      "references/dataframes.md",
      "references/futures.md",
      "references/schedulers.md"
    ]
  },
  "content": "Dask provides five main components, each suited to different use cases:\r\n\r\n### 1. DataFrames - Parallel Pandas Operations\r\n\r\n**Purpose**: Scale pandas operations to larger datasets through parallel processing.\r\n\r\n**When to Use**:\r\n- Tabular data exceeds available RAM\r\n- Need to process multiple CSV/Parquet files together\r\n- Pandas operations are slow and need parallelization\r\n- Scaling from pandas prototype to production\r\n\r\n**Reference Documentation**: For comprehensive guidance on Dask DataFrames, refer to `references/dataframes.md` which includes:\r\n- Reading data (single files, multiple files, glob patterns)\r\n- Common operations (filtering, groupby, joins, aggregations)\r\n- Custom operations with `map_partitions`\r\n- Performance optimization tips\r\n- Common patterns (ETL, time series, multi-file processing)\r\n\r\n**Quick Example**:\r\n```python\r\nimport dask.dataframe as dd\r\n\r\nddf = dd.read_csv('data/2024-*.csv')\r\n\r\nfiltered = ddf[ddf['value'] > 100]\r\nresult = filtered.groupby('category').mean().compute()\r\n```\r\n\r\n**Key Points**:\r\n- Operations are lazy (build task graph) until `.compute()` called\r\n- Use `map_partitions` for efficient custom operations\r\n- Convert to DataFrame early when working with structured data from other sources\r\n\r\n### 2. Arrays - Parallel NumPy Operations\r\n\r\n**Purpose**: Extend NumPy capabilities to datasets larger than memory using blocked algorithms.\r\n\r\n**When to Use**:\r\n- Arrays exceed available RAM\r\n- NumPy operations need parallelization\r\n- Working with scientific datasets (HDF5, Zarr, NetCDF)\r\n- Need parallel linear algebra or array operations\r\n\r\n**Reference Documentation**: For comprehensive guidance on Dask Arrays, refer to `references/arrays.md` which includes:\r\n- Creating arrays (from NumPy, random, from disk)\r\n- Chunking strategies and optimization\r\n- Common operations (arithmetic, reductions, linear algebra)\r\n- Custom operations with `map_blocks`\r\n- Integration with HDF5, Zarr, and XArray\r\n\r\n**Quick Example**:\r\n```python\r\nimport dask.array as da\r\n\r\nx = da.random.random((100000, 100000), chunks=(10000, 10000))\r\n\r\ny = x + 100\r\nz = y.mean(axis=0)\r\n\r\nresult = z.compute()\r\n```\r\n\r\n**Key Points**:\r\n- Chunk size is critical (aim for ~100 MB per chunk)\r\n- Operations work on chunks in parallel\r\n- Rechunk data when needed for efficient operations\r\n- Use `map_blocks` for operations not available in Dask\r\n\r\n### 3. Bags - Parallel Processing of Unstructured Data\r\n\r\n**Purpose**: Process unstructured or semi-structured data (text, JSON, logs) with functional operations.\r\n\r\n**When to Use**:\r\n- Processing text files, logs, or JSON records\r\n- Data cleaning and ETL before structured analysis\r\n- Working with Python objects that don't fit array/dataframe formats\r\n- Need memory-efficient streaming processing\r\n\r\n**Reference Documentation**: For comprehensive guidance on Dask Bags, refer to `references/bags.md` which includes:\r\n- Reading text and JSON files\r\n- Functional operations (map, filter, fold, groupby)\r\n- Converting to DataFrames\r\n- Common patterns (log analysis, JSON processing, text processing)\r\n- Performance considerations\r\n\r\n**Quick Example**:\r\n```python\r\nimport dask.bag as db\r\nimport json\r\n\r\nbag = db.read_text('logs/*.json').map(json.loads)\r\n\r\nvalid = bag.filter(lambda x: x['status'] == 'valid')\r\nprocessed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})\r\n\r\nddf = processed.to_dataframe()\r\n```\r\n\r\n**Key Points**:\r\n- Use for initial data cleaning, then convert to DataFrame/Array\r\n- Use `foldby` instead of `groupby` for better performance\r\n- Operations are streaming and memory-efficient\r\n- Convert to structured formats (DataFrame) for complex operations\r\n\r\n### 4. Futures - Task-Based Parallelization\r\n\r\n**Purpose**: Build custom parallel workflows with fine-grained control over task execution and dependencies.\r\n\r\n**When to Use**:\r\n- Building dynamic, evolving workflows\r\n- Need immediate task execution (not lazy)\r\n- Computations depend on runtime conditions\r\n- Implementing custom parallel algorithms\r\n- Need stateful computations\r\n\r\n**Reference Documentation**: For comprehensive guidance on Dask Futures, refer to `references/futures.md` which includes:\r\n- Setting up distributed client\r\n- Submitting tasks and working with futures\r\n- Task dependencies and data movement\r\n- Advanced coordination (queues, locks, events, actors)\r\n- Common patterns (parameter sweeps, dynamic tasks, iterative algorithms)\r\n\r\n**Quick Example**:\r\n```python\r\nfrom dask.distributed import Client\r\n\r\nclient = Client()  # Create local cluster\r\n\r\ndef process(x):\r\n    return x ** 2\r\n\r\nfutures = client.map(process, range(100))\r\n\r\nresults = client.gather(futures)\r\n\r\nclient.close()\r\n```\r\n\r\n**Key Points**:\r\n- Requires distributed client (even for single machine)\r\n- Tasks execute immediately when submitted\r\n- Pre-scatter large data to avoid repeated transfers\r\n- ~1ms overhead per task (not suitable for millions of tiny tasks)\r\n- Use actors for stateful workflows\r\n\r\n### 5. Schedulers - Execution Backends\r\n\r\n**Purpose**: Control how and where Dask tasks execute (threads, processes, distributed).\r\n\r\n**When to Choose Scheduler**:\r\n- **Threads** (default): NumPy/Pandas operations, GIL-releasing libraries, shared memory benefit\r\n- **Processes**: Pure Python code, text processing, GIL-bound operations\r\n- **Synchronous**: Debugging with pdb, profiling, understanding errors\r\n- **Distributed**: Need dashboard, multi-machine clusters, advanced features\r\n\r\n**Reference Documentation**: For comprehensive guidance on Dask Schedulers, refer to `references/schedulers.md` which includes:\r\n- Detailed scheduler descriptions and characteristics\r\n- Configuration methods (global, context manager, per-compute)\r\n- Performance considerations and overhead\r\n- Common patterns and troubleshooting\r\n- Thread configuration for optimal performance\r\n\r\n**Quick Example**:\r\n```python\r\nimport dask\r\nimport dask.dataframe as dd\r\n\r\nddf = dd.read_csv('data.csv')\r\nresult1 = ddf.mean().compute()  # Uses threads\r\n\r\nimport dask.bag as db\r\nbag = db.read_text('logs/*.txt')\r\nresult2 = bag.map(python_function).compute(scheduler='processes')\r\n\r\ndask.config.set(scheduler='synchronous')\r\nresult3 = problematic_computation.compute()  # Can use pdb\r\n\r\n\r\nFor comprehensive performance optimization guidance, memory management strategies, and common pitfalls to avoid, refer to `references/best-practices.md`. Key principles include:\r\n\r\n### Start with Simpler Solutions\r\nBefore using Dask, explore:\r\n- Better algorithms\r\n- Efficient file formats (Parquet instead of CSV)\r\n- Compiled code (Numba, Cython)\r\n- Data sampling\r\n\r\n### Critical Performance Rules\r\n\r\n**1. Don't Load Data Locally Then Hand to Dask**\r\n```python\r\nimport pandas as pd\r\ndf = pd.read_csv('large.csv')\r\nddf = dd.from_pandas(df, npartitions=10)\r\n\r\nimport dask.dataframe as dd\r\nddf = dd.read_csv('large.csv')\r\n```\r\n\r\n**2. Avoid Repeated compute() Calls**\r\n```python\r\nfor item in items:\r\n    result = dask_computation(item).compute()\r\n\r\n\r\n### ETL Pipeline\r\n```python\r\nimport dask.dataframe as dd\r\n\r\nddf = dd.read_csv('raw_data/*.csv')\r\n\r\nddf = ddf[ddf['status'] == 'valid']\r\nddf['amount'] = ddf['amount'].astype('float64')\r\nddf = ddf.dropna(subset=['important_col'])\r\n\r\nsummary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})\r\nsummary.to_parquet('output/summary.parquet')\r\n```\r\n\r\n### Unstructured to Structured Pipeline\r\n```python\r\nimport dask.bag as db\r\nimport json\r\n\r\nbag = db.read_text('logs/*.json').map(json.loads)\r\nbag = bag.filter(lambda x: x['status'] == 'valid')\r\n\r\nddf = bag.to_dataframe()\r\nresult = ddf.groupby('category').mean().compute()\r\n```\r\n\r\n### Large-Scale Array Computation\r\n```python\r\nimport dask.array as da\r\n\r\nx = da.from_zarr('large_dataset.zarr')\r\n\r\nnormalized = (x - x.mean()) / x.std()\r\n\r\nda.to_zarr(normalized, 'normalized.zarr')\r\n```\r\n\r\n### Custom Parallel Workflow\r\n```python\r\nfrom dask.distributed import Client\r\n\r\nclient = Client()\r\n\r\ndata = client.scatter(large_dataset)\r\n\r\nfutures = []\r\nfor param in parameters:\r\n    future = client.submit(process, data, param)\r\n    futures.append(future)\r\n\r\n\r\n### File Formats\r\n- **Efficient**: Parquet, HDF5, Zarr (columnar, compressed, parallel-friendly)\r\n- **Compatible but slower**: CSV (use for initial ingestion only)\r\n- **For Arrays**: HDF5, Zarr, NetCDF\r\n\r\n### Conversion Between Collections\r\n```python\r\nddf = bag.to_dataframe()\r\n\r\narr = ddf.to_dask_array(lengths=True)\r\n\r\n\r\n### Iterative Development Workflow\r\n\r\n1. **Test on small data with synchronous scheduler**:\r\n```python\r\ndask.config.set(scheduler='synchronous')\r\nresult = computation.compute()  # Can use pdb, easy debugging\r\n```\r\n\r\n2. **Validate with threads on sample**:\r\n```python\r\nsample = ddf.head(1000)  # Small sample",
  "name": "dask",
  "id": "scientific-pkg-dask",
  "sections": {
    "Integration Considerations": "ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])\r\n```\r\n\r\n### With Other Libraries\r\n- **XArray**: Wraps Dask arrays with labeled dimensions (geospatial, imaging)\r\n- **Dask-ML**: Machine learning with scikit-learn compatible APIs\r\n- **Distributed**: Advanced cluster management and monitoring",
    "Best Practices": "computations = [dask_computation(item) for item in items]\r\nresults = dask.compute(*computations)\r\n```\r\n\r\n**3. Don't Build Excessively Large Task Graphs**\r\n- Increase chunk sizes if millions of tasks\r\n- Use `map_partitions`/`map_blocks` to fuse operations\r\n- Check task graph size: `len(ddf.__dask_graph__())`\r\n\r\n**4. Choose Appropriate Chunk Sizes**\r\n- Target: ~100 MB per chunk (or 10 chunks per core in worker memory)\r\n- Too large: Memory overflow\r\n- Too small: Scheduling overhead\r\n\r\n**5. Use the Dashboard**\r\n```python\r\nfrom dask.distributed import Client\r\nclient = Client()\r\nprint(client.dashboard_link)  # Monitor performance, identify bottlenecks\r\n```",
    "Reference Files": "All reference documentation files can be read as needed for detailed information:\r\n\r\n- `references/dataframes.md` - Complete Dask DataFrame guide\r\n- `references/arrays.md` - Complete Dask Array guide\r\n- `references/bags.md` - Complete Dask Bag guide\r\n- `references/futures.md` - Complete Dask Futures and distributed computing guide\r\n- `references/schedulers.md` - Complete scheduler selection and configuration guide\r\n- `references/best-practices.md` - Comprehensive performance optimization and troubleshooting\r\n\r\nLoad these files when users need detailed information about specific Dask components, operations, or patterns beyond the quick guidance provided here.",
    "Selecting the Right Component": "Use this decision guide to choose the appropriate Dask component:\r\n\r\n**Data Type**:\r\n- Tabular data → **DataFrames**\r\n- Numeric arrays → **Arrays**\r\n- Text/JSON/logs → **Bags** (then convert to DataFrame)\r\n- Custom Python objects → **Bags** or **Futures**\r\n\r\n**Operation Type**:\r\n- Standard pandas operations → **DataFrames**\r\n- Standard NumPy operations → **Arrays**\r\n- Custom parallel tasks → **Futures**\r\n- Text processing/ETL → **Bags**\r\n\r\n**Control Level**:\r\n- High-level, automatic → **DataFrames/Arrays**\r\n- Low-level, manual → **Futures**\r\n\r\n**Workflow Type**:\r\n- Static computation graph → **DataFrames/Arrays/Bags**\r\n- Dynamic, evolving → **Futures**",
    "Overview": "Dask is a Python library for parallel and distributed computing that enables three critical capabilities:\r\n- **Larger-than-memory execution** on single machines for data exceeding available RAM\r\n- **Parallel processing** for improved computational speed across multiple cores\r\n- **Distributed computation** supporting terabyte-scale datasets across multiple machines\r\n\r\nDask scales from laptops (processing ~100 GiB) to clusters (processing ~100 TiB) while maintaining familiar Python APIs.",
    "When to Use This Skill": "This skill should be used when:\r\n- Process datasets that exceed available RAM\r\n- Scale pandas or NumPy operations to larger datasets\r\n- Parallelize computations for performance improvements\r\n- Process multiple files efficiently (CSVs, Parquet, JSON, text logs)\r\n- Build custom parallel workflows with task dependencies\r\n- Distribute workloads across multiple cores or machines",
    "Debugging and Development": "```\r\n\r\n3. **Scale with distributed for monitoring**:\r\n```python\r\nfrom dask.distributed import Client\r\nclient = Client()\r\nprint(client.dashboard_link)  # Monitor performance\r\nresult = computation.compute()\r\n```\r\n\r\n### Common Issues\r\n\r\n**Memory Errors**:\r\n- Decrease chunk sizes\r\n- Use `persist()` strategically and delete when done\r\n- Check for memory leaks in custom functions\r\n\r\n**Slow Start**:\r\n- Task graph too large (increase chunk sizes)\r\n- Use `map_partitions` or `map_blocks` to reduce tasks\r\n\r\n**Poor Parallelization**:\r\n- Chunks too large (increase number of partitions)\r\n- Using threads with Python code (switch to processes)\r\n- Data dependencies preventing parallelism",
    "Core Capabilities": "from dask.distributed import Client\r\nclient = Client()\r\nresult4 = computation.compute()  # Uses distributed with dashboard\r\n```\r\n\r\n**Key Points**:\r\n- Threads: Lowest overhead (~10 µs/task), best for numeric work\r\n- Processes: Avoids GIL (~10 ms/task), best for Python work\r\n- Distributed: Monitoring dashboard (~1 ms/task), scales to clusters\r\n- Can switch schedulers per computation or globally",
    "Common Workflow Patterns": "results = client.gather(futures)\r\n```"
  }
}

---
name: dask
description: "Parallel/distributed computing. Scale pandas/NumPy beyond memory, parallel DataFrames/Arrays, multi-file processing, task graphs, for larger-than-RAM datasets and parallel workflows."
---

# Dask

## Overview

Dask is a Python library for parallel and distributed computing that enables three critical capabilities:
- **Larger-than-memory execution** on single machines for data exceeding available RAM
- **Parallel processing** for improved computational speed across multiple cores
- **Distributed computation** supporting terabyte-scale datasets across multiple machines

Dask scales from laptops (processing ~100 GiB) to clusters (processing ~100 TiB) while maintaining familiar Python APIs.

## When to Use This Skill

This skill should be used when:
- Process datasets that exceed available RAM
- Scale pandas or NumPy operations to larger datasets
- Parallelize computations for performance improvements
- Process multiple files efficiently (CSVs, Parquet, JSON, text logs)
- Build custom parallel workflows with task dependencies
- Distribute workloads across multiple cores or machines

## Core Capabilities

Dask provides five main components, each suited to different use cases:

### 1. DataFrames - Parallel Pandas Operations

**Purpose**: Scale pandas operations to larger datasets through parallel processing.

**When to Use**:
- Tabular data exceeds available RAM
- Need to process multiple CSV/Parquet files together
- Pandas operations are slow and need parallelization
- Scaling from pandas prototype to production

**Reference Documentation**: For comprehensive guidance on Dask DataFrames, refer to `references/dataframes.md` which includes:
- Reading data (single files, multiple files, glob patterns)
- Common operations (filtering, groupby, joins, aggregations)
- Custom operations with `map_partitions`
- Performance optimization tips
- Common patterns (ETL, time series, multi-file processing)

**Quick Example**:
```python
import dask.dataframe as dd

# Read multiple files as single DataFrame
ddf = dd.read_csv('data/2024-*.csv')

# Operations are lazy until compute()
filtered = ddf[ddf['value'] > 100]
result = filtered.groupby('category').mean().compute()
```

**Key Points**:
- Operations are lazy (build task graph) until `.compute()` called
- Use `map_partitions` for efficient custom operations
- Convert to DataFrame early when working with structured data from other sources

### 2. Arrays - Parallel NumPy Operations

**Purpose**: Extend NumPy capabilities to datasets larger than memory using blocked algorithms.

**When to Use**:
- Arrays exceed available RAM
- NumPy operations need parallelization
- Working with scientific datasets (HDF5, Zarr, NetCDF)
- Need parallel linear algebra or array operations

**Reference Documentation**: For comprehensive guidance on Dask Arrays, refer to `references/arrays.md` which includes:
- Creating arrays (from NumPy, random, from disk)
- Chunking strategies and optimization
- Common operations (arithmetic, reductions, linear algebra)
- Custom operations with `map_blocks`
- Integration with HDF5, Zarr, and XArray

**Quick Example**:
```python
import dask.array as da

# Create large array with chunks
x = da.random.random((100000, 100000), chunks=(10000, 10000))

# Operations are lazy
y = x + 100
z = y.mean(axis=0)

# Compute result
result = z.compute()
```

**Key Points**:
- Chunk size is critical (aim for ~100 MB per chunk)
- Operations work on chunks in parallel
- Rechunk data when needed for efficient operations
- Use `map_blocks` for operations not available in Dask

### 3. Bags - Parallel Processing of Unstructured Data

**Purpose**: Process unstructured or semi-structured data (text, JSON, logs) with functional operations.

**When to Use**:
- Processing text files, logs, or JSON records
- Data cleaning and ETL before structured analysis
- Working with Python objects that don't fit array/dataframe formats
- Need memory-efficient streaming processing

**Reference Documentation**: For comprehensive guidance on Dask Bags, refer to `references/bags.md` which includes:
- Reading text and JSON files
- Functional operations (map, filter, fold, groupby)
- Converting to DataFrames
- Common patterns (log analysis, JSON processing, text processing)
- Performance considerations

**Quick Example**:
```python
import dask.bag as db
import json

# Read and parse JSON files
bag = db.read_text('logs/*.json').map(json.loads)

# Filter and transform
valid = bag.filter(lambda x: x['status'] == 'valid')
processed = valid.map(lambda x: {'id': x['id'], 'value': x['value']})

# Convert to DataFrame for analysis
ddf = processed.to_dataframe()
```

**Key Points**:
- Use for initial data cleaning, then convert to DataFrame/Array
- Use `foldby` instead of `groupby` for better performance
- Operations are streaming and memory-efficient
- Convert to structured formats (DataFrame) for complex operations

### 4. Futures - Task-Based Parallelization

**Purpose**: Build custom parallel workflows with fine-grained control over task execution and dependencies.

**When to Use**:
- Building dynamic, evolving workflows
- Need immediate task execution (not lazy)
- Computations depend on runtime conditions
- Implementing custom parallel algorithms
- Need stateful computations

**Reference Documentation**: For comprehensive guidance on Dask Futures, refer to `references/futures.md` which includes:
- Setting up distributed client
- Submitting tasks and working with futures
- Task dependencies and data movement
- Advanced coordination (queues, locks, events, actors)
- Common patterns (parameter sweeps, dynamic tasks, iterative algorithms)

**Quick Example**:
```python
from dask.distributed import Client

client = Client()  # Create local cluster

# Submit tasks (executes immediately)
def process(x):
    return x ** 2

futures = client.map(process, range(100))

# Gather results
results = client.gather(futures)

client.close()
```

**Key Points**:
- Requires distributed client (even for single machine)
- Tasks execute immediately when submitted
- Pre-scatter large data to avoid repeated transfers
- ~1ms overhead per task (not suitable for millions of tiny tasks)
- Use actors for stateful workflows

### 5. Schedulers - Execution Backends

**Purpose**: Control how and where Dask tasks execute (threads, processes, distributed).

**When to Choose Scheduler**:
- **Threads** (default): NumPy/Pandas operations, GIL-releasing libraries, shared memory benefit
- **Processes**: Pure Python code, text processing, GIL-bound operations
- **Synchronous**: Debugging with pdb, profiling, understanding errors
- **Distributed**: Need dashboard, multi-machine clusters, advanced features

**Reference Documentation**: For comprehensive guidance on Dask Schedulers, refer to `references/schedulers.md` which includes:
- Detailed scheduler descriptions and characteristics
- Configuration methods (global, context manager, per-compute)
- Performance considerations and overhead
- Common patterns and troubleshooting
- Thread configuration for optimal performance

**Quick Example**:
```python
import dask
import dask.dataframe as dd

# Use threads for DataFrame (default, good for numeric)
ddf = dd.read_csv('data.csv')
result1 = ddf.mean().compute()  # Uses threads

# Use processes for Python-heavy work
import dask.bag as db
bag = db.read_text('logs/*.txt')
result2 = bag.map(python_function).compute(scheduler='processes')

# Use synchronous for debugging
dask.config.set(scheduler='synchronous')
result3 = problematic_computation.compute()  # Can use pdb

# Use distributed for monitoring and scaling
from dask.distributed import Client
client = Client()
result4 = computation.compute()  # Uses distributed with dashboard
```

**Key Points**:
- Threads: Lowest overhead (~10 µs/task), best for numeric work
- Processes: Avoids GIL (~10 ms/task), best for Python work
- Distributed: Monitoring dashboard (~1 ms/task), scales to clusters
- Can switch schedulers per computation or globally

## Best Practices

For comprehensive performance optimization guidance, memory management strategies, and common pitfalls to avoid, refer to `references/best-practices.md`. Key principles include:

### Start with Simpler Solutions
Before using Dask, explore:
- Better algorithms
- Efficient file formats (Parquet instead of CSV)
- Compiled code (Numba, Cython)
- Data sampling

### Critical Performance Rules

**1. Don't Load Data Locally Then Hand to Dask**
```python
# Wrong: Loads all data in memory first
import pandas as pd
df = pd.read_csv('large.csv')
ddf = dd.from_pandas(df, npartitions=10)

# Correct: Let Dask handle loading
import dask.dataframe as dd
ddf = dd.read_csv('large.csv')
```

**2. Avoid Repeated compute() Calls**
```python
# Wrong: Each compute is separate
for item in items:
    result = dask_computation(item).compute()

# Correct: Single compute for all
computations = [dask_computation(item) for item in items]
results = dask.compute(*computations)
```

**3. Don't Build Excessively Large Task Graphs**
- Increase chunk sizes if millions of tasks
- Use `map_partitions`/`map_blocks` to fuse operations
- Check task graph size: `len(ddf.__dask_graph__())`

**4. Choose Appropriate Chunk Sizes**
- Target: ~100 MB per chunk (or 10 chunks per core in worker memory)
- Too large: Memory overflow
- Too small: Scheduling overhead

**5. Use the Dashboard**
```python
from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitor performance, identify bottlenecks
```

## Common Workflow Patterns

### ETL Pipeline
```python
import dask.dataframe as dd

# Extract: Read data
ddf = dd.read_csv('raw_data/*.csv')

# Transform: Clean and process
ddf = ddf[ddf['status'] == 'valid']
ddf['amount'] = ddf['amount'].astype('float64')
ddf = ddf.dropna(subset=['important_col'])

# Load: Aggregate and save
summary = ddf.groupby('category').agg({'amount': ['sum', 'mean']})
summary.to_parquet('output/summary.parquet')
```

### Unstructured to Structured Pipeline
```python
import dask.bag as db
import json

# Start with Bag for unstructured data
bag = db.read_text('logs/*.json').map(json.loads)
bag = bag.filter(lambda x: x['status'] == 'valid')

# Convert to DataFrame for structured analysis
ddf = bag.to_dataframe()
result = ddf.groupby('category').mean().compute()
```

### Large-Scale Array Computation
```python
import dask.array as da

# Load or create large array
x = da.from_zarr('large_dataset.zarr')

# Process in chunks
normalized = (x - x.mean()) / x.std()

# Save result
da.to_zarr(normalized, 'normalized.zarr')
```

### Custom Parallel Workflow
```python
from dask.distributed import Client

client = Client()

# Scatter large dataset once
data = client.scatter(large_dataset)

# Process in parallel with dependencies
futures = []
for param in parameters:
    future = client.submit(process, data, param)
    futures.append(future)

# Gather results
results = client.gather(futures)
```

## Selecting the Right Component

Use this decision guide to choose the appropriate Dask component:

**Data Type**:
- Tabular data → **DataFrames**
- Numeric arrays → **Arrays**
- Text/JSON/logs → **Bags** (then convert to DataFrame)
- Custom Python objects → **Bags** or **Futures**

**Operation Type**:
- Standard pandas operations → **DataFrames**
- Standard NumPy operations → **Arrays**
- Custom parallel tasks → **Futures**
- Text processing/ETL → **Bags**

**Control Level**:
- High-level, automatic → **DataFrames/Arrays**
- Low-level, manual → **Futures**

**Workflow Type**:
- Static computation graph → **DataFrames/Arrays/Bags**
- Dynamic, evolving → **Futures**

## Integration Considerations

### File Formats
- **Efficient**: Parquet, HDF5, Zarr (columnar, compressed, parallel-friendly)
- **Compatible but slower**: CSV (use for initial ingestion only)
- **For Arrays**: HDF5, Zarr, NetCDF

### Conversion Between Collections
```python
# Bag → DataFrame
ddf = bag.to_dataframe()

# DataFrame → Array (for numeric data)
arr = ddf.to_dask_array(lengths=True)

# Array → DataFrame
ddf = dd.from_dask_array(arr, columns=['col1', 'col2'])
```

### With Other Libraries
- **XArray**: Wraps Dask arrays with labeled dimensions (geospatial, imaging)
- **Dask-ML**: Machine learning with scikit-learn compatible APIs
- **Distributed**: Advanced cluster management and monitoring

## Debugging and Development

### Iterative Development Workflow

1. **Test on small data with synchronous scheduler**:
```python
dask.config.set(scheduler='synchronous')
result = computation.compute()  # Can use pdb, easy debugging
```

2. **Validate with threads on sample**:
```python
sample = ddf.head(1000)  # Small sample
# Test logic, then scale to full dataset
```

3. **Scale with distributed for monitoring**:
```python
from dask.distributed import Client
client = Client()
print(client.dashboard_link)  # Monitor performance
result = computation.compute()
```

### Common Issues

**Memory Errors**:
- Decrease chunk sizes
- Use `persist()` strategically and delete when done
- Check for memory leaks in custom functions

**Slow Start**:
- Task graph too large (increase chunk sizes)
- Use `map_partitions` or `map_blocks` to reduce tasks

**Poor Parallelization**:
- Chunks too large (increase number of partitions)
- Using threads with Python code (switch to processes)
- Data dependencies preventing parallelism

## Reference Files

All reference documentation files can be read as needed for detailed information:

- `references/dataframes.md` - Complete Dask DataFrame guide
- `references/arrays.md` - Complete Dask Array guide
- `references/bags.md` - Complete Dask Bag guide
- `references/futures.md` - Complete Dask Futures and distributed computing guide
- `references/schedulers.md` - Complete scheduler selection and configuration guide
- `references/best-practices.md` - Comprehensive performance optimization and troubleshooting

Load these files when users need detailed information about specific Dask components, operations, or patterns beyond the quick guidance provided here.