Skip to content

Shuffle storage and checkpointing

Some operations in a distributed query need data held by other workers, so the query transfers data between the workers. This transfer is a shuffle. The workers store the shuffle data while the query runs. Its location has a large impact on query execution times, so benchmark the options on the target infrastructure before choosing one.

  • Worker-local disk, the default: each worker serves its shuffle files to the other workers. The data does not survive a worker failure.
  • A shared filesystem or object storage holds the data independently of the workers and survives a worker failure, so checkpointing is not needed. Performance depends on the latency and throughput of the storage relative to local volumes.

Worker-local disk

With worker-local storage, workers write shuffle data to a file, preferably on a worker-local SSD. Other workers read it through the worker that wrote it:

worker_1 -[fs write]-> disk
worker_2 <-[net]- worker_1 <-[fs read]- disk

By default, each worker pod stores shuffle data in an emptyDir volume. To keep the host from running out of disk space during large queries, enable a generic ephemeral volume per worker pod instead:

shuffleData:
  ephemeralVolumeClaim:
    enabled: true
    storageClassName: <STORAGE_CLASS>

Set worker.shuffle_location.local.path to a directory on the worker's local disk:

[worker]
enabled = true
shuffle_location.local.path = "/mnt/storage/polars/shuffle"
# ...

Shared filesystem

A shared filesystem such as NFS or CephFS can hold the shuffle data too. Workers write directly to the shared filesystem, and any worker reads directly from it. All workers must use the same path.

worker_1 -[net]-> shared storage -[fs]-> disk
worker_2 <-[net]- shared storage <-[fs]- disk

Enable shuffleData.sharedFilesystem and mount the volume at its path, /mnt/nfs/shuffle-data by default, with worker.deployment.volumes and worker.deployment.runtimeContainer.volumeMounts:

shuffleData:
  sharedFilesystem:
    enabled: true

worker:
  deployment:
    volumes:
      - name: nfs-shuffle
        nfs:
          server: <NFS_SERVER>
          path: /exports/shuffle-data
    runtimeContainer:
      volumeMounts:
        - name: nfs-shuffle
          mountPath: /mnt/nfs/shuffle-data

With a ReadWriteMany storage class, the chart can create and mount a shared volume instead.

shuffleData:
  sharedPersistentVolumeClaim:
    enabled: true
    storageClassName: <RWX_STORAGE_CLASS>

[worker]
enabled = true
shuffle_location.shared_filesystem.path = "/mnt/storage/polars/shuffle"
# ...

Object storage

With object storage, every worker writes its shuffle partitions to the same prefix, and any worker reads them back. S3, Google Cloud Storage and Azure Blob Storage are supported. Any other provider that supports the S3 API, such as MinIO or DigitalOcean Spaces, works as well. For self-hosted S3-compatible storage, set the aws_endpoint_url option. The storage options take the same keys as the storage_options of scan_parquet(), such as aws_access_key_id, google_service_account_path or azure_storage_account_name.

Enable one of shuffleData.s3, shuffleData.gcs or shuffleData.abs, with the full URI in endpoint and the storage options in options, as literal values or from a Secret:

shuffleData:
  s3:
    enabled: true
    endpoint: "s3://<BUCKET>/<PREFIX>"
    options:
      - name: aws_access_key_id
        valueFrom:
          secretKeyRef:
            name: <SECRET_NAME>
            key: accessKeyId
      - name: aws_secret_access_key
        valueFrom:
          secretKeyRef:
            name: <SECRET_NAME>
            key: secretAccessKey

[worker]
enabled = true
shuffle_location.s3.url = "s3://<BUCKET>/<PREFIX>"
shuffle_location.s3.aws_access_key_id = "<AWS_ACCESS_KEY_ID>"
shuffle_location.s3.aws_secret_access_key = "<AWS_SECRET_ACCESS_KEY>"

The [worker] section of the config file reference lists the gcs and abs keys.

Checkpointing

Checkpointing persists the intermediate state of a long distributed query while it runs. If a worker fails part way through, for example when its executor is killed (see Out-of-memory behavior), the query resumes from the last checkpoint instead of starting over. Checkpointing is a no-op when the shuffle data already goes to a shared filesystem or object storage: the intermediate state is then persisted there as part of normal execution.

The checkpoint location is a shared filesystem or object storage that is durable and that every worker can reach, with the same providers and storage options as shuffle data in object storage. There is no worker-local variant: a checkpoint must survive the loss of the worker that wrote it.

Checkpointing is enabled automatically when a checkpointData location is configured: s3, gcs or abs with the full URI in endpoint and the storage options in options, a sharedFilesystem path on a volume you mount yourself under worker.deployment, or a sharedPersistentVolumeClaim that the chart creates and mounts with a ReadWriteMany storage class.

checkpointData:
  s3:
    enabled: true
    endpoint: "s3://<BUCKET>/<PREFIX>"
    options:
      - name: aws_access_key_id
        valueFrom:
          secretKeyRef:
            name: <SECRET_NAME>
            key: accessKeyId
      - name: aws_secret_access_key
        valueFrom:
          secretKeyRef:
            name: <SECRET_NAME>
            key: secretAccessKey

Enable checkpointing on the scheduler with the [scheduler.checkpoint] section, and set worker.checkpoint_location on every worker. period is required, and the config file reference lists its formats.

[scheduler.checkpoint]
period = "10 mins"

[worker]
enabled = true
checkpoint_location.s3.url = "s3://<BUCKET>/<PREFIX>"
checkpoint_location.s3.aws_access_key_id = "<AWS_ACCESS_KEY_ID>"
checkpoint_location.s3.aws_secret_access_key = "<AWS_SECRET_ACCESS_KEY>"

The [worker] section of the config file reference lists the shared_filesystem, gcs and abs keys.