# Inside the Pipeline: Fetching, Cleaning, and Embedding Survey Data at Scale

In [Part 6](https://blog.lucrolearning.com/making-workers-production-ready) we closed the loop on the execution platform: a worker process boots in isolation, reports progress through a callback, listens for `SIGTERM`, and cleans up after itself no matter how it exits. At that point we had a fully production-ready way to *run arbitrary long-lived work* but we never looked at what that work actually was.

It's time to open that box. Everything from here on happens inside a single method call the worker makes once it has loaded the pipeline class:

```python
analysis: PipeLineInterface = _load_analysis(analysis_path, constructor_args)
await analysis.run(progress_cb, stop_event=stop_event)
```

That one call `ThematicAnalysisPipeline.run()` is ten stages deep. It takes a batch of raw, free-text responses and turns them into clustered, summarized, mapped themes. This post covers the first six stages: getting the right slice of responses out of MongoDB, cleaning them, and turning them into embeddings without paying to re-embed anything twice. Part 8 will pick up from there and cover clustering, LLM summarization, and the mapping stage along with the resumability logic that lets a killed job pick up mid-clustering instead of starting over.

![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/87cc8edd-83a6-4208-98f3-7a5a8be2df9b.webp align="center")

Every stage below reports a milestone through the same `progress_cb` we introduced in Part 5 `"data fetched"`, `"data prune completed"`, `"embeddings loaded"`, and so on. That's not just a progress bar for the frontend; each milestone also resets the watchdog's staleness clock from Part 4. A stage that takes forty minutes to embed ten thousand responses is fine. A stage that goes forty minutes *without reporting anything* looks exactly like a hung process, and the watchdog will kill it. Keep that in mind as we go it's why almost every stage below is chunked and batched rather than done in one giant pass.

## Step 1: Fetching the Right Slice of Data

The pipeline doesn't operate on an entire dataset at once. It operates on one field within one dataset, seen through one caller's access scope. `fetch_data` delegates to a helper that builds a MongoDB query against the response records and pulls out just the answers for the target field:

```python
query = {"datasetId": dataset_id}

if filters:
    and_conditions = []
    for field_id, values in filters.items():
        and_conditions.append({
            "answers": {
                "$elemMatch": {
                    "fieldId": field_id,
                    "values": {"$in": values[0].split("_")}
                }
            }
        })
    query["$and"] = and_conditions

records = await db_client.fetch_many(
    "responses", query, {"answers": 1, "_id": 1}
)
```

Two details here are easy to miss but matter a lot in practice:

**Multi-value answers get split into pseudo-records.** An open-ended field like "what could we improve?" often comes back as a single comma-separated blob *"better pay, more flexible hours, less micromanagement."* Rather than treating that as one response, the pipeline splits it and gives each fragment its own synthetic ID (`{response_id}_{index}`), so three distinct themes don't get flattened into one incoherent response during clustering later. The split isn't naive string-splitting on commas, either a small guard list of conjunctions (`NO_SPLIT_WORDS`: "and", "because", "however", "for example"...) prevents cutting a sentence like *"pay is fine, but the hours are brutal"* into two meaningless halves.

**Hierarchy-scoped access is a BFS, not a flat lookup.** When a dataset has organizational-hierarchy scoping enabled, a small resolver walks down from the caller's scope through a parent-scope field on each record, breadth-first, collecting every descendant scope before the main query runs. Someone scoped to one business unit should see that unit and everything beneath it but the query can't express "everything beneath it" directly, so the pipeline has to compute the full set of scopes first and filter against it.

![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/2bb6f177-dde7-432e-90ad-0a569864f175.jpg align="center")

The output isn't a list of strings , it's a `pandas.DataFrame` with `response_id`, `response`, and `redacted_response` columns, plus the raw set of `response_ids`. That set matters more than it looks: it's the key the next two stages use to avoid redoing work.

## Step 2: Pruning Garbage Before It Costs You Anything

Free-text fields attract noise: `"n/a"`, `"...."`, `"xxxxx"`, single characters, and the occasional string of keyboard mashing. None of that is worth an embedding call or a clustering pass, so `prune_data` filters it out early before any paid API call touches it.

`DataFramePruner._is_meaningful` runs each response through a small stack of cheap filters, ordered from cheapest to most expensive:

```python
def _is_meaningful(self, value: str) -> bool:
    v = value.strip()

    if self.pattern.match(v):          # "na", "n/a", "tbd", "--", single letters...
        return False
    if self.gibberish_pattern.match(v):  # "aaaa", "!!!!", repeated chars
        return False

    if self.ai_detector:
        result = self.ai_detector(v[:512], top_k=None)
        gibberish_score = next((r['score'] for r in result if r['label'].lower() == 'noise'), 0.0)
        if gibberish_score > 0.8:
            return False

    if self.use_nlp_filter:
        meaningful_words = [t for t in tokens if t not in self.stopwords]
        if len(meaningful_words) == 0:   # response is *only* stopwords
            return False

    return True
```

The two regexes catch the obvious junk for free. The optional AI detector a locally-loaded gibberish-classification model, only wired up if `AI_DETECTOR_PATH` is set , it catches subtler noise the regexes miss.

The stopword check catches the residual case where a response is grammatically valid but semantically empty, like *"it is what it is."*

Pruning isn't just cosmetic it's a gate. If fewer than 50 responses survive, the pipeline stops outright and marks the run `INSUFFICIENT_DATA` instead of running an entire clustering pipeline on a handful of data points:

```python
if len(df_pruned) < 50:
    await self.update_state(self.dataset_id, self.field_id, "INSUFFICIENT_DATA")
    return None, response_info
```

> **Key insight:** every expensive stage downstream: embedding, clustering, LLM refinement costs real money and real time per response. Pruning aggressively at the top of the funnel isn't an optimization, it's what makes the economics of the rest of the pipeline work.

## Step 3: Don't Pay to Embed the Same Response Twice

This is the stage I'd call the most architecturally interesting of the six, and it's easy to walk right past it. Before generating a single new embedding, the pipeline checks what it's already embedded for this exact dataset/field pair, using the `response_ids` collected back in Step 1:

```python
async def fetch_with_retry(chunk):
    async with semaphore:
        return await self.client.fetch_many(
            "embeddings_new",
            {"datasetId": self.dataset_id, "fieldId": self.field_id,
             "responseId": {"$in": chunk}},
            {"_id": 0, "responseId": 1, "processedResponse": 1, "embedding": 1}
        )

tasks = [fetch_with_retry(value_list[start:start + batch_size])
          for start in range(0, len(value_list), batch_size)]
results = await asyncio.gather(*tasks)
```

Response IDs are chunked into batches of 1,000, fetched concurrently under a semaphore (so a dataset with fifty thousand responses doesn't fire fifty thousand-ID queries or fifty unbounded ones), with exponential backoff plus jitter on failure. The result is two dictionaries `existing_responses` and `existing_embedding_map` keyed by response ID.

`split_new_existing` then does exactly what it sounds like: any response already found gets routed into `df_existing` (its previously-computed clean text gets reattached), and everything else goes into `df_new`; the only rows that will actually touch the embedding API in Step 5.

![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/f1541ea0-8448-4669-ac6c-9ea22d5f3d8b.jpg align="center")

This is a cache lookup, functionally, but it's worth naming as a distinct architectural decision rather than an implementation detail. Datasets get re-analyzed as new responses trickle in over time. Without this step, every re-run would re-embed every response ever collected, and embedding cost would grow without bound instead of scaling with *new* responses only.

## Step 4: Normalizing Text for the Model

Only `df_new` the responses that survived pruning and aren't already embedded goes through `TextPreprocessor`. The interesting design choice here is that the preprocessing steps are configuration, not code:

```python
self.step_funcs = {
    "remove_placeholder": self._remove_placeholders,
    "lowercase": self._lowercase,
    "remove_punctuation": self._remove_punctuation,
    "remove_whitespace": self._remove_whitespace,
    "remove_stopwords": self._remove_stopwords,
    "lemmatize": self._lemmatize,
}
```

`PREPROCESS_CONFIG["steps"]` is just a list of strings naming which functions to run and in what order:

```python
"steps": ["remove_placeholder", "lowercase", "remove_punctuation",
          "remove_whitespace", "lemmatize", "remove_stopwords"]
```

`transform()` walks that list and calls `self.step_funcs[step]` for each name, batch by batch. Want to drop lemmatization for a faster run, or reorder stopword removal to happen before lowercasing? That's a config change, not a code change ; the same "swap behavior without touching the class" instinct we saw with `PIPELINE_CLASS_PATH` dynamic loading back in Part 5, just applied one level down.

![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/d141a32f-5852-4279-805b-70d19a2c9674.jpg align="center")

Lemmatization is the expensive step it runs every response through spaCy. The preprocessor checks for a GPU up front and falls back to spreading the work across CPU cores if theisn't one:

```python
if self.use_gpu_if_available and torch.cuda.is_available():
    spacy.require_gpu()
    n_process = 1
else:
    spacy.require_cpu()
    n_process = multiprocessing.cpu_count() - 1

self.nlp = spacy.load(self.spacy_model, disable=["ner", "parser"])
```

Note the `disable=["ner", "parser"]` the pipeline only needs lemmas and part-of-speech tags, so named-entity recognition and dependency parsing (two of spaCy's moexpensive components) aswitched off entirely. It's a small line, but on a dataset with tens of thousands of responses it's the difference between a preprocessing pass that takes seconds and one that takes minutes.

## Step 5: Generating Embeddings Without Melting the API

Only truly new responses reach this stage, and it reuses a pattern this codebase clearly likes: a thin interface with swappable providers.

```python
class BaseEmbeddingProvider(ABC):
    @abstractmethod
    def embed_data(self, texts: List[str]) -> List[List[float]]:
        pass
```

```python
def get_embedding_provider(provider: str, model_type: str = None, **kwargs):
    provider = provider.lower()
    if provider == "openai":
        return OpenAIEmbedding(model=model_type, **kwargs)
    elif provider == "vertex":
        return VertexEmbedding(model=model_type, **kwargs)
    elif provider == "huggingface":
        return HuggingFaceEmbedding(model_type=model_type)
    elif provider == "bedrock":
        return BedrockEmbedding(model_id=model_type, **kwargs)
```

This is the same shape as the `PipeLineInterface` / dynamic-import pattern from Part 5, just scoped to one concern: which embedding backend to call. Config picks `openai` / `text-embedding-3-large` today; swapping to a self-hosted HuggingFace model for cost reasons is a config change, not a rewrite of the pipeline.

![](https://cdn.hashnode.com/uploads/covers/6a51387ca8cf56c623690316/59e6c039-37ef-45f5-b4c0-fc7ef01a0d83.jpg align="center")

The orchestration around the actual API calls is where the concurrency discipline from earlier posts shows up again this time applied to an external, rate-limited API instead of internal processes:

```python
async def process_chunk(start_idx, end_idx):
    chunk_raw = raw_responses[start_idx:end_idx]
    async with self.api_semaphore:
        embeddings = await asyncio.to_thread(
            self.embedding_service.generate_embeddings, chunk_raw
        )
    async with self.db_semaphore:
        await self.db_service.save_embeddings(embeddings, ..., collection_name)
    return embeddings

tasks = [process_chunk(i, min(i + self.chunk_size, len(raw_responses)))
          for i in range(0, len(raw_responses), self.chunk_size)]
chunk_results = await asyncio.gather(*tasks)
```

Responses are split into chunks of 200, and two independent semaphores bound the work: one caps how many embedding API calls are in flight at once, the other caps concurrent MongoDB writes. They're deliberately separate the API and the database have completely different capacity limits, and coupling them would mean tuning for whichever one is more fragile.

`EmbeddingService.generate_embeddings` adds its own retry loop with backoff on top of this so a transient rate-limit error on one chunk doesn't take down the whole batch, it just delays that one chunk's `asyncio.gather` slot.

## Step 6: Recombining Old and New

The last stage in this post is the shortest, but it's where the two paths from Step 3 rejoin:

```python
def combine_data_for_clustering(self, df_new_processed, df_existing, new_embedding_map, existing_embedding_map):
    df_all = pd.concat([df_new_processed, df_existing], ignore_index=True)

    embedding_vectors = [
        new_embedding_map[rid] if rid in new_embedding_map else existing_embedding_map.get(rid)
        for rid in df_all["response_id"]
    ]

    df_all['redacted_response'] = df_all['redacted_response'].combine_first(df_all["response"])
    df_all.rename(columns={"redacted_response": "response"}, inplace=True)

    return df_all, embedding_vectors
```

Two things worth calling out. First, the embedding vector list is rebuilt by looking up each response ID in *either* map this is the point where "responses we just embedded" and "responses we already had embeddings for" become a single aligned dataset, ready to hand to clustering as one `responses`/`embeddings` pair. Second, and easy to miss: `combine_first` means the **redacted** version of a response wins whenever one exists, falling back to the raw response only when no redaction was recorded. Every downstream stage clustering, LLM refinement, summarization sees the PII-safe version of the text, not the original.

## What to Notice Across All Six Steps

Look back across fetch → prune → dedupe → preprocess → embed → combine and a few patterns repeat on purpose:

*   **Provider abstraction shows up twice** (embeddings here, LLMs in Part 8) using the same "interface + factory function" shape as the dynamic pipeline loading from Part 5. It's the same idea applied at three different layers of the system.
    
*   **Every I/O-bound stage is chunked and semaphore-bounded** response ID lookups, embedding API calls, and DB writes all cap their own concurrency independently, because each has a different real-world capacity limit.
    
*   **Config, not code, controls behavior** preprocessing steps, batch sizes, chunk sizes, and provider choice all live in plain dictionaries in `pipelines/config/`, not scattered through the pipeline logic.
    
*   **Cheap filters run before expensive ones** regex before AI-model inference in pruning; a cache lookup before any paid API call in the embedding stage.
    

None of this is exotic engineering. It's the boring, disciplined kind bound your concurrency, cache what you can, gate expensive work behind cheap checks applied consistently at every stage. That discipline is exactly what makes it possible for a job to survive a `SIGTERM` mid-embedding and pick back up later without corrupting state or double-billing an API.

That survivability becomes the whole story in Part 8, where we cover the three stages that actually produce a "theme": clustering, LLM-based topic refinement, map-reduce summarization across an arbitrary number of themes, and the resume logic that lets the pipeline recognize "clustering already finished, just re-enter at the mapping stage" instead of starting from `fetch_data` again.