Quickstart

Minimal package

acme_steps/uppercase/
  uppercase.drtml
  hooks.py
  processor.py

Manifest

drtml_version: 3
step_id: uppercase
execution:
  mode: partition_loop
  entry: acme_steps.uppercase.hooks:run

params_defaults:
  runtime.document_batch_size: 32
  runtime.metrics_enabled: true

storage:
  bindings:
    artifacts: {backend: local, root: /tmp/drtoller-data}

datasets:
  documents:
    role: input
    binding: artifacts
    columns: {document_id: string, text: string}
    feed:
      loop: primary
      read: {mode: batch, batch_rows: 256}
      partition: {source: file}

  uppercase_documents:
    role: output
    binding: artifacts
    backend: parquet
    columns: {run_id: string, document_id: string, text: string}

processing:
  processor: acme_steps.uppercase.processor:process_batch
  batch_processor: acme_steps.uppercase.processor:process_batch
  outputs:
    documents: uppercase_documents

Processor

from typing import Any, Mapping
from drtoller.framework.processing.process_result import ProcessResult
from drtoller.framework.storage.feed.batch import ProcessorBatch


def process_batch(
    raws: list[Any], *, ctx, params: Mapping[str, Any],
    processor_ctx: Mapping[str, Any], partition_id: str,
) -> list[ProcessResult]:
    del params, processor_ctx, partition_id
    results = []
    for batch in raws:
        if not isinstance(batch, ProcessorBatch):
            raise TypeError("expected ProcessorBatch")
        rows = [
            {
                "run_id": ctx.run_id,
                "document_id": str(row["document_id"]),
                "text": str(row.get("text") or "").upper(),
            }
            for row in batch.primary_rows
        ]
        results.append(ProcessResult(ok=True, rows={"documents": rows}, shard_id=batch.shard_id))
    return results

documents is a logical output key. DRTML maps it to physical dataset uppercase_documents.

Hook

from drtoller.framework.drtml.resolve import manifest_path_adjacent
from drtoller.framework.processing.partition_loop import run_partition_loop


def run(ctx, *, manifest_path=None):
    return run_partition_loop(
        ctx,
        manifest_path=manifest_path or manifest_path_adjacent(__file__),
    )

Framework reads the feed, invokes the processor, writes rows, manages flush/checkpoints and emits declared metrics.

Next