AllenNeuralDynamics / AllenNeuralDynamics/analysis-pipeline-utils

Dispatcher job-skipping does one DocDB query per job; ~40 min for 2000 jobs

Ouverte
#59 1 commentaire 0 réactions 0 personnes assignées Voir sur GitHub
Langage dominant
Python
Étoiles
0
Forks
1
Métriques de merge des PR
Aucune PR mergée en 30 j

Description

A production pipeline running for years accumulates a large number of already-completed jobs that have to be skipped on every dispatch. My prototype pipeline is already past 275K, and the current dispatcher does one DocDB round trip per job, so the skip check scales with work that has already been done rather than with work remaining.

Concretely: on the new framework, job-skipping took **~40 minutes for 2000 jobs** (~1.2 s/job). See [this run](https://codeocean.allenneuraldynamics.org/capsule/8929013/tree?results=6830592d-fe25-4ceb-aded-524bcd025eac_computation).

## For comparison: my prototype pipeline

From a recent run ([capsule 7444014](https://codeocean.allenneuraldynamics.org/capsule/7444014/tree?results=c32280d6-3831-4245-b18d-a14a48b3f612_computation)):

```
08:08:48 - Generated 157838 total jobs.
08:08:48 - Retrieving existing job hashes from docDB...
08:10:11 - Found 275880 existing job hashes in docDB.
08:10:11 - Assigned pending 25 jobs to 25 workers.
08:10:11 - 157813 already existed.
```

157,838 jobs checked against 275,880 existing records in **~83 seconds total** — 83 s for one bulk fetch, then **0.13 s** to classify all 157,838 jobs.

At the new pipeline's ~1.2 s/job, those same 157,838 jobs would take **50+ hours**.

## Why the difference

The prototype does one paginated fetch of `_id` and then a set difference in memory ([job-manager `docDB_io.py`](https://github.com/AllenNeuralDynamics/aind-analysis-arch-job-manager/blob/main/code/util/docDB_io.py)):

```python
records = analysis_docDB_dft.retrieve_docdb_records(
filter_query={},
projection={"_id": 1},
paginate=True,
paginate_batch_size=100000,
)
```

```python
new_job_hashes = list(set(all_job_hashes) - set(existing_job_hashes))
```

One network round trip regardless of job count. The current dispatcher instead queries per job — `check_task_parameters` (`utils_analysis_dispatch.py:457`) calls `docdb_record_exists(process.code)` inside the loop, which goes to `get_docdb_records` (`metadata.py:523`) and issues `retrieve_docdb_records(filter_query={"name": docdb_id})`, one query per job.

Note `max_number_of_tasks_dispatched` doesn't limit this. The generator stops once enough *new* jobs are found, so in steady-state production — where nearly everything is already done — it still walks essentially the whole backlog first. The bad case is the normal case.

## Proposals

**1. Query `_id` instead of `name`.** `_id` is set to the same `processing_prefix()` hash as `name` (`result_files.py:63`), and it's indexed by DocDB automatically, whereas `name` presumably isn't. So every per-job lookup today is likely scanning a growing collection, which would explain ~1.2 s/job and means the per-query cost keeps degrading over time independently of job count. This is a small change and probably the single biggest win.

**2. Fetch existing hashes once and compare in memory.** The structural fix, as in the prototype: one paginated fetch with `projection={"_id": 1}`, build a `set`, test membership locally. `DOCDB_COLLECTION` is already per-analysis, so the fetch is naturally bounded. This turns O(N) round trips into one, and is what gets us from hours to seconds.

The two compose — 1 makes the remaining query cheap, 2 removes almost all of them.

I also looked at parallelizing the job-file writes, but the numbers point at DocDB round trips rather than local disk I/O, so I don't think that's where the time is going.

Guide de contribution

Aucun guide de contribution indexé pour ce dépôt

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.