The Complete Production SOP: How to Deploy and Run Dask Distributed DataFrame & Parallel Tasks in Autonomous AI Agents

---
name: "dask-distributed-dataframe"
description: "Scale Pandas DataFrames across multi-core CPUs and distributed compute clusters with Dask."
metadata:
  author: "Dask"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "use dask dataframe"
  - "scale pandas across cores"
  - "run distributed dask cluster"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Dask Distributed DataFrame & Parallel Tasks

Parallelizes Python data science workloads using Dask: Dask DataFrame, delayed task graphs, distributed Client clusters, partition optimization, and chunked Parquet writing.

## When to Invoke
Invoke this skill whenever you need to:
- use dask dataframe
- scale pandas across cores
- run distributed dask cluster

## Execution SOP & Rules
1. Initialize cluster connection using from dask.distributed import Client; client = Client().
2. Read partitioned datasets with dd.read_parquet('data/*.parquet').
3. Target partition sizes between 100MB and 250MB using df.repartition(partition_size='150MB').
4. Trigger computation only when necessary with df.compute() to avoid premature memory consumption.

## Verification Checklist
- Run local linting, syntax validation, and type-checks before reporting completion.
- Ensure all created or modified files adhere to workspace boundaries and avoid uncommitted secrets.
- Verify tests or build commands exit cleanly with return code 0.

import dask

# Enforce deterministic worker thresholds to avoid OOM crashes
dask.config.set({
    'distributed.worker.memory.target': 0.60,
    'distributed.worker.memory.spill': 0.70,
    'distributed.worker.memory.pause': 0.80,
    'distributed.worker.memory.terminate': 0.95,
    'dataframe.backend': 'pyarrow'
})

from dask.distributed import Client
import dask.dataframe as dd
import pipeline_config  # Applies Dask memory thresholds

def run_pipeline(input_path: str, output_path: str):
    # Initialize local cluster matching physical core count
    client = Client(processes=True, silence_logs=True)
    
    try:
        # Lazy read via columnar engine
        df = dd.read_parquet(input_path, engine='pyarrow')
        
        # Lazy aggregation pipeline
        aggregated = (
            df.groupby(['merchant_id', 'transaction_date'])
            .agg({'amount': ['sum', 'count', 'mean']})
        )
        aggregated.columns = ['revenue_sum', 'tx_count', 'avg_ticket']
        aggregated = aggregated.reset_index()
        
        # Maintain recommended 100MB-250MB partition sizing
        optimized_df = aggregated.repartition(partition_size='150MB')
        
        # Persist chunked results directly to storage; avoid client-side compute()
        optimized_df.to_parquet(output_path, engine='pyarrow', write_index=False)
    finally:
        client.close()

if __name__ == '__main__':
    run_pipeline('./raw_data/*.parquet', './processed/')

import pytest
from dask.distributed import Client
import dask.dataframe as dd
import pandas as pd
from process_transactions import run_pipeline

def test_dask_pipeline_output(tmp_path):
    # Setup minimal mock dataset
    input_dir = tmp_path / 'raw'
    output_dir = tmp_path / 'out'
    input_dir.mkdir()
    
    mock_data = pd.DataFrame({
        'merchant_id': [101, 102, 101, 103],
        'transaction_date': ['2026-03-01', '2026-03-01', '2026-03-02', '2026-03-02'],
        'amount': [50.0, 120.0, 75.5, 300.0]
    })
    mock_data.to_parquet(input_dir / 'data.parquet')
    
    run_pipeline(str(input_dir / '*.parquet'), str(output_dir))
    
    result = dd.read_parquet(str(output_dir)).compute()
    assert len(result) == 3
    assert 'revenue_sum' in result.columns
    assert result.loc[result['merchant_id'] == 101, 'tx_count'].iloc[0] == 1