Python Dask中persist与scatter的区别及适用场景咨询

Core Differences #

scatter() #

• Purpose: Explicitly sends small, client-side in-memory objects to Dask workers. These objects are stored in worker memory and made available to tasks.
• Data type: Works with non-Dask objects (e.g., Python lists, small pandas DataFrames, dictionaries) that exist on the client machine.
• Mechanism: Transfers data directly from the client to workers. The client must hold the entire object in memory before sending.

persist() #

• Purpose: Triggers computation of a lazy Dask collection (like dask.DataFrame, dask.Array) and stores the resulting partitions in worker memory. It converts a lazy collection into a "persisted" version where future operations reuse the in-memory partitions instead of recomputing from scratch.
• Data type: Designed for large Dask collections defined via lazy operations (e.g., reading from cloud storage, filtering, aggregating).
• Mechanism: Computes the collection in parallel across workers. Each worker processes and stores its assigned partition directly, without the client needing to hold the entire dataset in memory.

Best Fit Scenarios #

When to use scatter() #

• You have a small object (megabytes, not gigabytes) that multiple tasks need access to (e.g., a lookup table, hyperparameters for a model).
• You need explicit control over sharing client-side data with workers for specific tasks.
• Example:

  lookup_table = pd.read_csv("small_lookup.csv")
  scattered_lookup = client.scatter(lookup_table, broadcast=True)
  # Use scattered_lookup in tasks that need this table
  

When to use persist() #

• You’re working with large TB-scale datasets and need to reuse the same collection across multiple computations (e.g., running multiple queries on the same filtered DataFrame).
• You want to avoid recomputing expensive operations (like reading from slow storage or complex transformations) every time you run a new task.
• Example:

  ddf = dask.dataframe.read_parquet("s3://large-dataset/*.parquet")
  filtered_ddf = ddf[ddf["value"] > 100]
  persisted_ddf = filtered_ddf.persist()  # Computes and stores partitions in worker memory
  # Subsequent operations on persisted_ddf use in-memory data
  

Do you need scatter() after persist()? #

No—persist() already handles storing your Dask collection’s partitions in worker memory. Future operations on the persisted collection will directly access these in-memory partitions without needing to transfer data from the client.

The only time you might use scatter() alongside persist() is if you have an additional small client-side object that tasks need to use with the persisted collection (e.g., a threshold value for filtering). But you never need to scatter the persisted collection itself.

Why your TB-scale dataset failed with scatter() #

When you tried to use scatter() for a TB-scale dataset, the client couldn’t hold the entire dataset in memory to send to workers—this is a fundamental limitation of scatter(), which is designed for small data. persist() solves this by distributing the computation across workers, each handling its own partition, so the client only manages the task graph (not the raw data).

内容的提问来源于stack exchange，提问作者Axel Wang