The foundation: Why Temporal for a data pipeline?

This is a guest post written by Houman Kargaran, Engineering Lead at ANZ Bank

In this series, I walk through building a PII-compliant complaint ingestion pipeline for a large enterprise — using Temporal as the orchestration layer. The full working code is available at https://github.com/houmanka/PII-Compliant-RAG.

The pipeline processes CSV files containing customer complaints, scans each row for PII, classifies it using an internal ML model, stores only the redacted text, and makes it searchable through semantic search. All of this happens without raw customer data leaving the enterprise infrastructure. No direct nor indirect PII should appear in the Temporal Web UI.

This is a series about why you would reach for Temporal when building a data pipeline under real engineering constraints: compliance requirements, no external API calls during processing, and a hard requirement that no complaint is ever silently lost.

Each article in the series focuses on one layer of the system. This first article begins where many projects do: the requirements and the design decision.

Objective#

This client is a large enterprise struggling to process a high volume of complaints and requests. It needs semantic search across that data. CSV files land in a cloud bucket, and because the complaints arrive unclassified, each one must be classified. Each row might have PII that must be redacted before the complaint is stored or embedded. The data cannot be shared with an external provider; it must remain within the enterprise.

Functional requirements#

The pipeline must:

Start processing when it receives a Pub/Sub event.
Retry recoverable errors encountered while processing complaint data.
Check each complaint for PII using an in-house MCP server.
Classify each complaint using an internal ML model.
Store only redacted complaint data.
Support semantic search using embeddings generated only from redacted complaints.
Use an external vector database in production only after the provider has been reviewed and approved by the compliance team.

Non-functional requirements:#

Resilience: The application must not lose any data and be resilient to failures.
Consistency: Every complaint must be stored in the database. None can be skipped.
Availability: If an instance shuts down, processing must continue from the last recorded point.
Observability: Every step of the pipeline must be visible and auditable.

Pipeline steps:#

Ingest file: The ingest_file function streams the file from the cloud bucket.
PII check: Each row is checked for PII and redaction.
Classification: The in-house classifier determines the complaint’s category.
Storage: The redacted text, case ID, and classification are stored in the database.
Embedding: The redacted complaints are embedded, and the embeddings are stored in the cache.
Vector DB: Pending approval by the compliance team. (This simply means you must code in a manner that can be swapped with ease. Pending approval is telling us Factory design pattern)
Sanity check: The unique cache ID is used to query the vector database and verify that the data was stored correctly.
Clean up: The cache is cleared, and the Workflow finishes.

What I have illustrated here is a sample of how I would prepare a design document. You may have noticed that I have not mentioned Temporal yet, and there is a good reason for that. I cannot simply use Temporal because I want to, even though other teams in the company already use it. In an enterprise, having another team get through the red tape and secure approval for a software application means much of the hard work is done. Now I need to justify why Temporal is a good fit for this project. You must address the functional and non-functional requirements thoroughly because a board of architects will read the document.

Naive approach#

Resilience: We need to persist state in Redis or a database so we can recover from a crash — which is hard to maintain.
Consistency: We need a relational database and must persist every step. We also need to have a strategy for safe retries and duplicate work.
Availability: We need a state machine so if an instance goes down, the Workers can resume from the last known good state. We also need to keep that state up to date.
Observability: We need to build a UI, which dramatically increases the scope and brings the frontend team into the project.

If we choose this approach, we need to change the requirements and possibly break the project down further.

Why Temporal?#

Resilience: Temporal provides Durable Execution and retries out of the box. If a Worker shuts down mid-file, the Workflow continues when a Worker is available again. In this implementation, the ingestion Activity heartbeats its progress, allowing a retry to resume from the last recorded row. Without Temporal, we would have to build and maintain that retry and resume logic ourselves.
Consistency: Temporal records completed Activity results in the Workflow’s Event History. Workflow code is deterministic, while Activity code remains responsible for idempotency because Activity Execution is at least once by default.
Availability: If a Worker process restarts or an instance goes down, the Workflow replays from its persisted Event History and continues. Its recorded progress is not lost.
Observability: The status of each Activity in the pipeline is visible in the Temporal Web UI. In a regulated environment, being able to show auditors which stages ran, when they ran, and whether they succeeded matters. Row-level audit details still require application-level logging or persistence.

Recommendation#

Based on the requirements, Temporal is the right orchestration layer for this pipeline.

Now that I have decided to use Temporal, I will go into more detail. In a live design document, I would expand a bit more on the naive approach. I have kept it short here for simplicity.

A brief architecture overview#

The pipeline starts when a CSV file lands in a cloud bucket. A Pub/Sub event is published and picked up by a lightweight subscriber, which invokes ingestion_handler as a Temporal Standalone Activity. This Activity starts ComplaintWorkflow, which orchestrates the full processing chain:

IngestFileActivity streams the CSV row by row, runs each row through the MCP PII scan, classifies it with the internal ML model, and persists only the redacted text to Postgres.
EmbeddingActivity fetches the unembedded complaints, generates 384-dimension vectors using all-MiniLM-L6-v2, and caches them in Redis keyed by a unique session ID.
VectorStorageActivity reads the cache and upserts the vectors into the vector database, then runs a sanity query to confirm the data landed correctly.
CacheActivity cleans up the Redis key to close the loop.

Every step is a discrete Temporal Activity with its own Retry Policy. No step re-runs unless it actually fails.

See the working diagrams for more details.

Mapping requirements to a Workflow#

This usually would not appear in the design document because, once it is approved, engineers may be held to every word. Still, I want to show how I am thinking as I write it.

@workflow.defn(name="ComplaintWorkflow")
class ComplaintWorkflow:
   @workflow.run
   # TODO: Ingest, PII MCP call, ML classifier
   # TODO: Store in DB
   async def run(self, file_input: FileInput) -> bool:
       file_id = await workflow.execute_activity(
           ingest_file_activity,
           FileDetails(path=file_input.path),
           start_to_close_timeout=timedelta(seconds=120),
           heartbeat_timeout=timedelta(seconds=20),
       )

       # TODO: use an approved model to embed
       # TODO: cache it internally
       embedding_result: EmbeddingActivityResult = await workflow.execute_activity(
           embedding_activity,
           file_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       # TODO some retry policy since, this needs to be external
       vector_storage: VectorStorageActivityResult = await workflow.execute_activity(
           store_vector,
           embedding_result.cache_id,
           start_to_close_timeout=timedelta(seconds=120),
           retry_policy=retry()
       )

       # TODO maybe think about logging the vector_storage

       # TODO avoid double process of the embedded records
       await workflow.execute_activity(
           update_embedded_records,
           file_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       # TODO: use what we have in the cache and do sanity check
       await workflow.execute_activity(
           query_vector,
           embedding_result.cache_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       # TODO: do the clean up
       await workflow.execute_activity(
           delete_cache,
           embedding_result.cache_id,
           start_to_close_timeout=timedelta(seconds=120),
       )

       workflow.logger.info("completed")
       return True

I generally map out what my Workflow will contain, the relationships between its parts, and some ideas about what comes next. This might not be important in a small company, but in an enterprise, changes come from every direction, especially on a large project involving hundreds of engineers. When I have a map in my head, I know what goes where, and I can tell my business analyst or lead exactly how things will work. With that map, my junior engineers or an LLM can follow a specific pattern when creating the Activities. To be continued.

A forward pointer#

This article covered the problem, the requirements, and why Temporal is the right tool for the job. The rest of the series goes deeper on each layer:

Part 2. Bridging Pub/Sub to Temporal with Standalone Activities
Part 3. Using Activity isolation as a security boundary
Part 4. Chaining Activities: From text to vectors