Processing S3 Objects at Scale with Hatchet

Introduction

Amazon S3 is a reliable-and-boring distributed object store with strong read-after-write consistency, making it an ideal backend for processing massive datasets at scale. However, coordinating workloads that actually leverage S3’s scale (without accruing a comically large cloud bill) is non-trivial.

This guide walks through building an embarrassingly parallel Hatchet pipeline for ingesting and processing hundreds of thousands of images stored across multiple regionally bound S3 buckets. This approach will be making extensive use of Hatchet’s concurrency control features, paired with aggressive fan-outs via dynamic child spawning, to parallelize the heck out of a data processing workflow while ensuring fair allocation of resources and idempotent processing of objects.

Scenario

For the purposes of this guide, we’ll constrain the system as follows:

• Each bucket corresponds to a separate class of objects.
• Bucket sizes are unevenly distributed.
• Objects are ephemeral. Once successfully processed, an object should be deleted.
• Data can be added or removed at any point, with a sizable chunk being front-loaded at startup.

Setup

AWS Credentials

This pipeline requires AWS access keys configured with the following IAM permissions:

Note: <your-bucket-prefix> should be scoped at runtime to the specific bucket prefix you are targeting.

These can be set via environment variables, which should be automatically picked up by the SDK client:

const s3 = new S3Client({ forcePathStyle: true });

s3 = boto3.client("s3")

Workflows

To ensure maximum parallelism, we treat the processing of each object within each bucket as an independent unit of work. Concretely, the execution sequence is:

1. Fetch buckets from S3.
2. Fetch objects within each bucket.
3. Process each object.
4. Delete each object.

We implement this as a double fan-out: one fan-out across buckets to spawn per-bucket polling tasks, and a second fan-out within each polling task to spawn one processing task per object. This lets us distribute both the listing and processing work across a horizontally scaled fleet of workers without any single worker becoming a bottleneck.

Define the Models

Start by defining the data models for our workflow inputs.

type ListObjectsInput = {
  bucket: string;
};

type ProcessObjectInput = {
  bucket: string;
  key: string;
};

class ListObjectsInput(BaseModel):
    bucket: str


class ProcessObjectInput(BaseModel):
    bucket: str
    key: str

Notice that we only have models defined for our input data, as the task naturally concludes upon object deletion.

Fetch Buckets

We start by fetching all relevant buckets from S3. For simplicity, this pipeline forgoes S3 Event Notifications in favor of periodic scanning via a cron trigger.

As the root of the pipeline, this task paginates across all buckets in our region, fanning out a child workflow per bucket. Since we’re just iterating across mini-batches/pages of buckets, there’s no meaningful work here to parallelize.

However, our strategy should still guard against overlapping executions of the same cron job. If S3 throttles us mid-scan and a run takes longer than the cron interval, the next tick should be rejected until the in-flight run finishes, rather than piling on more concurrent listing calls and making the throttling worse.

const fetchBucketsWorkflow = hatchet.workflow({
  name: 'fetch_s3_buckets',
  on: {
    cron: '* * * * *',
  },
  concurrency: {
    expression: "'singleton'",
    maxRuns: 1,
    limitStrategy: ConcurrencyLimitStrategy.CANCEL_NEWEST,
  },
});

fetch_buckets_workflow = hatchet.workflow(
    name="fetch_s3_buckets",
    on_crons=["* * * * *"],
    concurrency=ConcurrencyExpression(
        expression="'singleton'",
        max_runs=1,
        limit_strategy=ConcurrencyLimitStrategy.CANCEL_NEWEST,
    ),
)

Here, we set the concurrency key to the CEL expression 'singleton', a string literal that evaluates to the same value for every run. Combined with max_runs=1 and CANCEL_NEWEST, this rule funnels all runs through a single slot on a single worker: while one is in-flight, any new run triggered by the cron is immediately canceled.

The task body itself is responsible for gathering all relevant buckets from our region and spawning a new fetch_objects task from the fetch_s3_objects workflow for each:

fetchBucketsWorkflow.task({
  name: 'fetch_buckets',
  fn: async () => {
    for await (const page of paginateListBuckets(
      { client: s3, pageSize: 10 },
      { Prefix: BUCKET_PREFIX }
    )) {
      const items = (page.Buckets ?? [])
        .filter((bucket): bucket is { Name: string } => bucket.Name !== undefined)
        .map((bucket) => ({
          input: { bucket: bucket.Name },
          opts: {
            childKey: bucket.Name,
            additionalMetadata: { 'bucket-name': bucket.Name },
          },
        }));

      if (items.length > 0) {
        await fetchObjectsWorkflow.runManyNoWait(items);
      }
    }

    return {};
  },
});

@fetch_buckets_workflow.task()
async def fetch_buckets(input: EmptyModel, ctx: Context) -> dict[str, Any]:
    paginator = s3.get_paginator("list_buckets")
    pages = paginator.paginate(
        Prefix=BUCKET_PREFIX,
        PaginationConfig={"PageSize": 10},
    )

    for page in pages:
        items = [
            fetch_objects_workflow.create_bulk_run_item(
                input=ListObjectsInput(bucket=b["Name"]),
                child_key=b["Name"],
                additional_metadata={"bucket-name": b["Name"]},
            )
            for b in page.get("Buckets", [])
        ]
        if items:
            await fetch_objects_workflow.aio_run_many(items, wait_for_result=False)

    return {}

We use the bucket name as its child key for each spawn to ensure deduplication of downstream fetch_objects runs.

Fetch Objects

Next, we define a workflow responsible for polling a given S3 bucket for objects, before fanning out and assigning each object a processing task.

Similarly to Fetch Buckets, this workflow needs to ensure each bucket has at most one poller in-flight at a time, preventing overlaps via the CANCEL_NEWEST strategy and constraining to a single run. Here, however, we group by bucket name using the CEL expression input.bucket to enforce per-bucket concurrency.

Additionally, we define a global concurrency expression that limits how many bucket pollers can spin up simultaneously. Tasks exceeding this rule are enqueued using the GROUP_ROUND_ROBIN strategy and will execute when concurrent slots free up.

const fetchObjectsWorkflow = hatchet.workflow<ListObjectsInput>({
  name: 'fetch_s3_objects',
  concurrency: [
    {
      expression: 'input.bucket',
      maxRuns: 1,
      limitStrategy: ConcurrencyLimitStrategy.CANCEL_NEWEST,
    },
    {
      expression: "'constant'",
      maxRuns: MAX_CONCURRENT_BUCKET_POLLERS,
      limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
    },
  ],
});

fetch_objects_workflow = hatchet.workflow(
    name="fetch_s3_objects",
    input_validator=ListObjectsInput,
    concurrency=[
        ConcurrencyExpression(
            expression="input.bucket",
            max_runs=1,
            limit_strategy=ConcurrencyLimitStrategy.CANCEL_NEWEST,
        ),
        ConcurrencyExpression.from_int(MAX_CONCURRENT_BUCKET_POLLERS),
    ],
)

These two ConcurrencyExpression entries compose to ensure a single poller per bucket paired with a workflow-level limitation on total in-flight pollers.

The task itself paginates through the bucket’s objects and spawns a processing run per object:

fetchObjectsWorkflow.task({
  name: 'fetch_objects',
  fn: async (input: ListObjectsInput) => {
    for await (const page of paginateListObjectsV2(
      { client: s3, pageSize: 100 },
      { Bucket: input.bucket }
    )) {
      const items = (page.Contents ?? [])
        .filter((obj): obj is { Key: string } => obj.Key !== undefined)
        .map((obj) => ({
          input: { bucket: input.bucket, key: obj.Key },
          opts: {
            childKey: `${input.bucket}/${obj.Key}`,
          },
        }));

      if (items.length > 0) {
        await processObjectWorkflow.runManyNoWait(items);
      }
    }

    return {};
  },
});

@fetch_objects_workflow.task()
async def fetch_objects(input: ListObjectsInput, ctx: Context) -> dict[str, Any]:
    paginator = s3.get_paginator("list_objects_v2")
    pages = paginator.paginate(
        Bucket=input.bucket,
        PaginationConfig={"PageSize": 100},
    )

    for page in pages:
        items = [
            process_object_workflow.create_bulk_run_item(
                input=ProcessObjectInput(bucket=input.bucket, key=obj["Key"]),
                child_key=f"{input.bucket}/{obj['Key']}",
            )
            for obj in page.get("Contents", [])
        ]
        if items:
            await process_object_workflow.aio_run_many(items, wait_for_result=False)

    return {}

The child key here combines the bucket name and object key ({bucket}/{key}), guaranteeing that re-triggering the parent won’t spawn duplicate processing runs for the same object. S3 object keys are only unique within a specific bucket, so the bucket prefix is mandatory here.

Process Objects

Lastly, we define the processing workflow, which represents the bulk of the pipeline’s actual computational work. Recall from the previous sections that:

1. The distribution of objects per bucket is not uniform.
2. Processed objects are immediately deleted.

Rather than capping concurrency globally, we scope the limit per bucket using S3_WORKER_MAX_RUNS_PER_BUCKET. A global cap would allow the largest buckets to monopolize all processing slots. A per-bucket cap keeps throughput fair across the system regardless of individual bucket size. When a bucket hits its limit, additional runs are enqueued and dispatched round-robin via GROUP_ROUND_ROBIN as slots open.

The diagram above shows this steady state: work interleaves across buckets, the worker pool runs at partial capacity, and a backlog drains in round-robin order. Even when bucket-1 has the most pending objects, it doesn’t monopolize the workers.

There is no child-side deduplication on this workflow (no concurrency expression keyed on {bucket}/{object-key}). Because process-object runs are only ever spawned by fetch_objects (which already handles deduplication via child key at spawn time) a secondary concurrency expression would be redundant.

const processObjectWorkflow = hatchet.workflow<ProcessObjectInput>({
  name: 'process_object',
  concurrency: {
    expression: 'input.bucket',
    maxRuns: MAX_RUNS_PER_BUCKET,
    limitStrategy: ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
  },
});

process_object_workflow = hatchet.workflow(
    name="process_object",
    input_validator=ProcessObjectInput,
    concurrency=ConcurrencyExpression(
        expression="input.bucket",
        max_runs=MAX_RUNS_PER_BUCKET,
        limit_strategy=ConcurrencyLimitStrategy.GROUP_ROUND_ROBIN,
    ),
)

The final step is to download the object, process it, and safely delete it from the source bucket:

processObjectWorkflow.task({
  name: 'process_object',
  fn: async (input: ProcessObjectInput, ctx) => {
    let body: Buffer;

    try {
      const response = await s3.send(
        new GetObjectCommand({ Bucket: input.bucket, Key: input.key })
      );
      if (!response.Body) {
        return {};
      }
      body = Buffer.from(await response.Body.transformToByteArray());
    } catch (err) {
      if (err instanceof NoSuchKey || err instanceof NoSuchBucket) {
        await ctx.log(`skipping ${input.bucket}/${input.key}: not found`);
        return {};
      }
      throw err;
    }

    // TODO: actual image processing here

    await s3.send(new DeleteObjectCommand({ Bucket: input.bucket, Key: input.key }));
    return {};
  },
});

@process_object_workflow.task()
def process_object(input: ProcessObjectInput, ctx: Context) -> dict[str, Any]:
    buf = BytesIO()
    try:
        s3.download_fileobj(input.bucket, input.key, buf)
    except s3.exceptions.ClientError as e:
        if (code := e.response["Error"]["Code"]) in {
            "NoSuchKey",
            "NoSuchBucket",
            "404",
        }:
            ctx.log(f"skipping {input.bucket}/{input.key}: not found ({code})")
            return {}
        raise

    buf.seek(0)
    # TODO: actual image processing here

    s3.delete_object(Bucket=input.bucket, Key=input.key)
    return {}

This architecture provides an idempotent, effectively-exactly-once processing pipeline via two core mechanisms:

1. Idempotent runs: Each download_and_process task fetches, processes, then deletes the target object. If the object does not exist, the NoSuchKey / NoSuchBucket (404) branch catches it and returns a no-op. S3’s strong read-after-write consistency ensures that a “not found” response is a reliable completion signal.
2. Spawn-time deduplication: Concurrent duplicate runs for the exact same (bucket, object) are actively prevented by the parent’s child_key constraint, meaning the 404 no-op path operates purely as a safety net rather than the norm.

Register and start the worker

You can start the worker, which will begin polling all S3 buckets prefixed with S3_WORKER_BUCKET_PREFIX as follows:

S3_WORKER_BUCKET_PREFIX="bucket-" \
S3_WORKER_MAX_CONCURRENT_BUCKET_POLLERS=10 \
S3_WORKER_MAX_RUNS_PER_BUCKET=20 \
S3_WORKER_SLOTS=40 \
poetry run python -m examples.aws.s3.worker

S3_WORKER_BUCKET_PREFIX="bucket-" \
S3_WORKER_MAX_CONCURRENT_BUCKET_POLLERS=10 \
S3_WORKER_MAX_RUNS_PER_BUCKET=20 \
S3_WORKER_SLOTS=40 \
npx ts-node src/v1/examples/aws/s3/worker.ts

docker compose -f examples/aws/s3/docker-compose.yml up -d --wait

AWS_ENDPOINT_URL=http://localhost:4566
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_REGION=us-east-1

Testing

pytest examples/aws/s3/test_worker.py