Skip to content

Read and write cloud storage

Remote queries scan and sink cloud storage with the same scan_*/sink_* API as open-source Polars. The scans run on the workers, which authenticate to the storage themselves, so the data is not downloaded to the client.

Set up

The workers can only read and write what their identity has access to.

The compute nodes run under the stack's IAM role whose name contains IAMWorkerRole, which reads and writes S3 by default. To let them read other data sources, open IAM in the AWS console, find this role and adjust its permissions, for instance to narrow down the S3 access to certain buckets.

To assume a role in another AWS account, list the worker role in that role's trust policy and allow sts:AssumeRole on that role in the worker role's permissions. See the AWS cross-account role documentation.

Bind a Kubernetes service account to a cloud identity with EKS Pod Identity, GKE Workload Identity or AKS Workload ID, and run the scheduler and the workers as that service account:

scheduler:
  serviceAccount:
    name: <SERVICE_ACCOUNT>
worker:
  serviceAccount:
    name: <SERVICE_ACCOUNT>

On AKS, also set podLabels to azure.workload.identity/use: "true", so Workload ID injects the identity's token into the pods.

Give the worker machines credentials for the storage, or pass the credentials per scan in storage_options.

Scan S3, GCS and Azure

An S3 scan passes no storage_options:

import polars as pl

q = pl.scan_parquet("s3://<BUCKET>/<PREFIX>/")

A Google Cloud Storage scan passes only the project:

q = pl.scan_parquet(
    "gs://<BUCKET>/<PREFIX>/",
    storage_options={"project": "<PROJECT>"},
)

An Azure Blob Storage scan passes only the storage account name:

q = pl.scan_parquet(
    "az://<CONTAINER>/<PREFIX>/",
    storage_options={"account_name": "<STORAGE_ACCOUNT>"},
)

Submit q with .remote(ctx) as described in Run a remote query.

Store large inputs as Parquet or as many files: a large Parquet file is split by row group across the workers, but one CSV or NDJSON file is read by a single worker. A <COLUMN>=<VALUE>/ directory layout avoids a shuffle only when its outer directory levels are the group_by or join keys, in order.

Assume another IAM role

To use another IAM role for a read or write operation, pass the role ARN in the operation's credential_provider parameter, and Polars assumes that role for it. This needs the boto3 package on the client and, on a cluster you run, on the workers (see Use Python packages and plugins). This example scans a Parquet file from S3 with the assumed role:

import polars as pl
import polars_nanook as pn

ctx: pn.ComputeContext | pn.ClusterContext

lf = pl.scan_parquet(
    "s3://<BUCKET>/foo.parquet",
    credential_provider=pl.CredentialProviderAWS(
        assume_role={
            "RoleArn": "<ROLE_ARN>",
            "RoleSessionName": "AssumedRole",
        },
    ),
)
lf.remote(ctx).collect()