Dask DataFrames

Overview

Dask DataFrames enable parallel processing of large tabular data by distributing work across multiple pandas DataFrames. As described in the documentation, "Dask DataFrames are a collection of many pandas DataFrames" with identical APIs, making the transition from pandas straightforward.

Core Concept

A Dask DataFrame is divided into multiple pandas DataFrames (partitions) along the index:

Each partition is a regular pandas DataFrame
Operations are applied to each partition in parallel
Results are combined automatically

Key Capabilities

Scale

Process 100 GiB on a laptop
Process 100 TiB on a cluster
Handle datasets exceeding available RAM

Compatibility

Implements most of the pandas API
Easy transition from pandas code
Works with familiar operations

When to Use Dask DataFrames

Use Dask When:

Dataset exceeds available RAM
Computations require significant time and pandas optimization hasn't helped
Need to scale from prototype (pandas) to production (larger data)
Working with multiple files that should be processed together

Stick with Pandas When:

Data fits comfortably in memory
Computations complete in subseconds
Simple operations without custom .apply() functions
Iterative development and exploration

Reading Data

Dask mirrors pandas reading syntax with added support for multiple files:

Single File

import dask.dataframe as dd

# Read single file
ddf = dd.read_csv('data.csv')
ddf = dd.read_parquet('data.parquet')

Multiple Files

# Read multiple files using glob patterns
ddf = dd.read_csv('data/*.csv')
ddf = dd.read_parquet('s3://mybucket/data/*.parquet')

# Read with path structure
ddf = dd.read_parquet('data/year=*/month=*/day=*.parquet')

Optimizations

# Specify columns to read (reduces memory)
ddf = dd.read_parquet('data.parquet', columns=['col1', 'col2'])

# Control partitioning
ddf = dd.read_csv('data.csv', blocksize='64MB')  # Creates 64MB partitions

Common Operations

All operations are lazy until .compute() is called.

Filtering

# Same as pandas
filtered = ddf[ddf['column'] > 100]
filtered = ddf.query('column > 100')

Column Operations

# Add columns
ddf['new_column'] = ddf['col1'] + ddf['col2']

# Select columns
subset = ddf[['col1', 'col2', 'col3']]

# Drop columns
ddf = ddf.drop(columns=['unnecessary_col'])

Aggregations

# Standard aggregations work as expected
mean = ddf['column'].mean().compute()
sum_total = ddf['column'].sum().compute()
counts = ddf['category'].value_counts().compute()

GroupBy

# GroupBy operations (may require shuffle)
grouped = ddf.groupby('category')['value'].mean().compute()

# Multiple aggregations
agg_result = ddf.groupby('category').agg({
    'value': ['mean', 'sum', 'count'],
    'amount': 'sum'
}).compute()

Joins and Merges

# Merge DataFrames
merged = dd.merge(ddf1, ddf2, on='key', how='left')

# Join on index
joined = ddf1.join(ddf2, on='key')

Sorting

# Sorting (expensive operation, requires data movement)
sorted_ddf = ddf.sort_values('column')
result = sorted_ddf.compute()

Custom Operations

Apply Functions

To Partitions (Efficient):

# Apply function to entire partitions
def custom_partition_function(partition_df):
    # partition_df is a pandas DataFrame
    return partition_df.assign(new_col=partition_df['col1'] * 2)

ddf = ddf.map_partitions(custom_partition_function)

To Rows (Less Efficient):

# Apply to each row (creates many tasks)
ddf['result'] = ddf.apply(lambda row: custom_function(row), axis=1, meta=('result', 'float'))

Note: Always prefer map_partitions over row-wise apply for better performance.

Meta Parameter

When Dask can't infer output structure, specify the meta parameter:

# For apply operations
ddf['new'] = ddf.apply(func, axis=1, meta=('new', 'float64'))

# For map_partitions
ddf = ddf.map_partitions(func, meta=pd.DataFrame({
    'col1': pd.Series(dtype='float64'),
    'col2': pd.Series(dtype='int64')
}))

Lazy Evaluation and Computation

Lazy Operations

# These operations are lazy (instant, no computation)
filtered = ddf[ddf['value'] > 100]
aggregated = filtered.groupby('category').mean()
final = aggregated[aggregated['value'] < 500]

# Nothing has computed yet

Triggering Computation

# Compute single result
result = final.compute()

# Compute multiple results efficiently
result1, result2, result3 = dask.compute(
    operation1,
    operation2,
    operation3
)

Persist in Memory

# Keep results in distributed memory for reuse
ddf_cached = ddf.persist()

# Now multiple operations on ddf_cached won't recompute
result1 = ddf_cached.mean().compute()
result2 = ddf_cached.sum().compute()

Index Management

Setting Index

# Set index (required for efficient joins and certain operations)
ddf = ddf.set_index('timestamp', sorted=True)

Index Properties

Sorted index enables efficient filtering and joins
Index determines partitioning
Some operations perform better with appropriate index

Writing Results

To Files

# Write to multiple files (one per partition)
ddf.to_parquet('output/data.parquet')
ddf.to_csv('output/data-*.csv')

# Write to single file (forces computation and concatenation)
ddf.compute().to_csv('output/single_file.csv')

To Memory (Pandas)

# Convert to pandas (loads all data in memory)
pdf = ddf.compute()

Performance Considerations

Efficient Operations

Column selection and filtering: Very efficient
Simple aggregations (sum, mean, count): Efficient
Row-wise operations on partitions: Efficient with map_partitions

Expensive Operations

Sorting: Requires data shuffle across workers
GroupBy with many groups: May require shuffle
Complex joins: Depends on data distribution
Row-wise apply: Creates many tasks

Optimization Tips

1. Select Columns Early

# Better: Read only needed columns
ddf = dd.read_parquet('data.parquet', columns=['col1', 'col2'])

2. Filter Before GroupBy

# Better: Reduce data before expensive operations
result = ddf[ddf['year'] == 2024].groupby('category').sum().compute()

3. Use Efficient File Formats

# Use Parquet instead of CSV for better performance
ddf.to_parquet('data.parquet')  # Faster, smaller, columnar

4. Repartition Appropriately

# If partitions are too small
ddf = ddf.repartition(npartitions=10)

# If partitions are too large
ddf = ddf.repartition(partition_size='100MB')

Common Patterns

ETL Pipeline

import dask.dataframe as dd

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

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

# Aggregate
summary = ddf.groupby('category').agg({
    'amount': ['sum', 'mean'],
    'quantity': 'count'
})

# Write results
summary.to_parquet('output/summary.parquet')

Time Series Analysis

# Read time series data
ddf = dd.read_parquet('timeseries/*.parquet')

# Set timestamp index
ddf = ddf.set_index('timestamp', sorted=True)

# Resample (if available in Dask version)
hourly = ddf.resample('1H').mean()

# Compute statistics
result = hourly.compute()

Combining Multiple Files

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

# Process combined data
result = ddf.groupby('category')['value'].sum().compute()

Limitations and Differences from Pandas

Not All Pandas Features Available

Some pandas operations are not implemented in Dask:

Some string methods
Certain window functions
Some specialized statistical functions

Partitioning Matters

Operations within partitions are efficient
Cross-partition operations may be expensive
Index-based operations benefit from sorted index

Lazy Evaluation

Operations don't execute until .compute()
Need to be aware of computation triggers
Can't inspect intermediate results without computing

Debugging Tips

Inspect Partitions

# Get number of partitions
print(ddf.npartitions)

# Compute single partition
first_partition = ddf.get_partition(0).compute()

# View first few rows (computes first partition)
print(ddf.head())

Validate Operations on Small Data

# Test on small sample first
sample = ddf.head(1000)
# Validate logic works
# Then scale to full dataset
result = ddf.compute()

Check Dtypes

# Verify data types are correct
print(ddf.dtypes)

/dask

daskreferencesdataframes.md