dask / dask/distributed

Unmanaged Memory Leak with Large Parquet Files (Dask + Distributed)

Open
#8,377 6 comments 1 reaction 0 assignees View on GitHub
bug
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

**Describe the issue**:

Not sure if this requires a separate bug report from #8375, but just in case, I can get the same error with plain old dask + distributed if the parquet data doesn't fit in memory.

When I try to read in 30 parquet files (~150 MB each) into 2 worker nodes (`m5.large`, 8 GB RAM, 500 GB disk each), there is a runaway unmanaged memory process that consumes all of the memory and causes the dask worker to be restarted. Eventually, after crashing on each of the 4 workers (2 nodes, 2 vCPUs each), dask gives up.

The process that is consuming all of the memory: `/usr/bin/python3 -c from multiprocessing.spawn import spawn_main; spawn_main(tracker_fd=11, pipe_handle=17) --multiprocessing-fork`

I now know that that's the mark of a forked process in Python, but I'm unable to figure out where in the code this occurs.

**Minimal Complete Verifiable Example**:

```python
import sys
import dask
import dask.distributed
from dask.distributed import Client
import dask.dataframe as dd

print(f"connecting to {sys.argv[1]}")
client = Client(sys.argv[1])

manifest_files = ['s3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_0934fdc9-1bc7-4dae-a371-c3f58f3b31fc',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_7011b49a-bf6c-42d0-9852-85aa10a3ad37',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_8ca98d70-74c2-4d17-b059-83cd25e398c0',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_61265893-e992-4026-8b69-e60a4641dd10',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_d4318a64-7fc0-4b59-8f85-8b1790e72a70',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_4e2e19a8-b360-49fe-9b82-f66a1e23ad3e',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_c5a77760-f078-432c-aaf5-6d99fb0cee0c',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_db5f90f3-9fc0-4b86-98da-d840bb9b5423',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_2472c560-1113-448b-ae18-e23032d3f3d8',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_551944ef-47f4-475d-a575-e209ca1ee7b4',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_67569166-18f9-40bd-9845-20f069f8dc8a',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_8a81dca0-767d-43bf-b0aa-a087f98d41c5',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_c9dd6a02-0f92-4202-ba70-77fda02d4acf',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_9f094026-03f4-4ec2-a348-0498eeb6b04d',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_2c20c942-1e2c-4f37-86ec-4adce83edbea',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_237d6c26-8372-4584-ae15-9f693d2295a6',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_cf41cbf5-eb46-4bb9-b904-f1518affbefa',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_420f8434-5a71-4af8-91b5-d5364c941ded',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_f155616d-39bc-483e-b19a-56da9fbae685',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_ef8a7770-43b7-44bb-9dde-f92ed8faae9b',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_8aa7de20-91ee-4ada-8f48-c7a5b313572f',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_7c265580-ada5-4b94-b818-5cebdb4bb6c6',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_13562caf-4578-4e36-83fe-8b7a5eabc7e8',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_b0f80cd3-4aa5-40ac-a0c8-632c763f8969',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_2f1e190b-621c-4638-91f1-e961ba674714',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_5dcd43e4-67bd-41ba-b3cd-7e8f8da9b1f9',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_293c6916-4f22-4ca0-a240-078ebe48368b',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_80e8c47e-f254-47c4-80af-d744885e8174',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_6b1ba112-0545-4c3f-9321-dbf597d98bc1',
's3://my-bucket/results/0.12892323498922598/20231127_175506_00007_3735a_0b3dd1a8-72ef-4852-bf1d-f7091629d3b6']

try:
df = dd.read_parquet(manifest_files)

divisions = list(range(0, 10001))
df = df.set_index('agent', divisions=divisions)

dp = df.persist()

print(dp.count().compute())

finally:
client.close()
```

**Anything else we need to know?**:

I'm running this on AWS. The dask cluster is handmade (via boto3), where my architecting script runs `dask worker` and `dask scheduler` on everything. The cluster seems to be correct because all other operations work, and the cluster is able to handle a 52 GB CSV file just fine — it only fails with parquet.

**Environment**:

Dask version: 2023.11.0
Python version: 3.10.12
Operating System: Ubuntu 22.04.3 LTS
Install method (conda, pip, source): pip

Logs:
[worker1.log](https://github.com/dask/distributed/files/13495572/worker1.log),
[worker2.log](https://github.com/dask/distributed/files/13495573/worker2.log),
[scheduler.log](https://github.com/dask/distributed/files/13495582/scheduler.log)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.