[ghukill]

Note

This PR is stacked on #281.

Purpose and background context

This 2nd PR in the stacked PRs for TIMX-738 introduces the ability to perform transformations via parallel worker processes. While this is the most impactful change, there are also some more minor (but still impactful) performance changes to the alma source specifically.

Until now, Transmogrifier was a single process, sequentially processing application. The architecture was a Transformer class that defines a __next__() method, effectively making it an iterator. If one were to call next(my_transformer) you would get a DatasetRecord instance which is the containing object of a transformed record.

For most sources, where records to transform are low, this is inconsequential. Total runtime maybe seconds to low minutes. But for sources like dspace, more recently digitalcollections, and certainly alma the single process, sequential nature of Transmogrifier becomes a major barrier to speed. Speed is not everything... but Transmogrifier is slow enough that it prohibits making substantial changes to transformation logic because full runs are expensive and cumbersome to complete. This is a problem.

Thankfully, much of the architecture lends itself well to parallelization!

This PR reworks the Transformer __next__(), __iter__(), and transform() methods slightly to lend themselves to transforming via parallel processes. To be precise, we are talking processes and not threads, where most of the time spent is CPU bound and threads would not be a meaningful speed increase.

This matters, because cross-process parallelization in python requires objects that are pickable/serializable to pass between processes. This required shifting when and where we parse raw source record bytes into BeautifulSoup4 objects (Tag) or dictionaries. By shifting that into code that would be isolated per process, the __next__() method becomes more of an orchestrator than transformation and business logic. This is arguably also a good thing! We as developers shouldn't have to muck around in this method once; it's preferable that farm out to methods we do touch.

The net effect is a dramatic speedup. Where a transformation may have taken 60 minutes previously, if we run Transmogrifier with --workers 8 and parallelize across 8 workers, we'd see a runtime of about 7.5 minutes (60 / 8). As we'll see in the final PR there is still more room for significant performance improvements, but parallelization is a major one.

How can a reviewer manually see the effects of these changes?

1- Set Dev1 TIMDEX AWS credentials

2- Sync data from S3 locally to isolate network speed variables:

aws s3 sync \
s3://timdex-extract-dev-222053980223/timmy/exports/alma-subject-criticism-and-interpretation/ \
/tmp/timmy/exports/transmog-parallel-base

3- Run Transmogrifier with a single worker (default also if --workers is not provided):

uv run transform \
--source alma \
--input-parquet /tmp/timmy/exports/transmog-parallel-base \
--output-parquet /tmp/timmy/exports/parallel-workers-single \
--workers 1

On my machine, this process takes around 165 seconds, which for a subset containing 13,941 records works out to roughly 87 records/second:

2026-11-03 10:13:59,408 INFO transmogrifier.cli.main(): Running transform for source alma
...
2026-11-03 10:16:44,664 INFO transmogrifier.cli.main(): Total time to complete transform: 0:02:45.256166

4- Now, we'll run with 8 parallel workers:

uv run transform \
--source alma \
--input-parquet /tmp/timmy/exports/transmog-parallel-base \
--output-parquet /tmp/timmy/exports/parallel-workers-eight \
--workers 8

This completes in roughly 27 seconds, which works out to 516 records/second, overall a roughly 6x speedup:

2026-11-03 10:20:43,703 INFO transmogrifier.cli.main(): Running transform for source alma
2026-11-03 10:20:43,912 INFO transmogrifier.sources.transformer._iter_with_workers(): Transforming records with 8 worker processes
...
2026-11-03 10:21:11,583 INFO transmogrifier.cli.main(): Total time to complete transform: 0:00:27.879940

Extrapolating that out, where something may have taken 3 hours it would now take 22.5 minutes. Even before the next pass of performance improvements in the next PR, we can now perform runs over substantial amounts of records to preview changes we are making.

Includes new or updated dependencies?

YES

Changes expectations for external applications?

NO: at this time, if --workers is excluded, we default to a single process. However, we may choose to allow the pipeline lambdas to set --workers for the Transmogrifier jobs in the future. And/or, update Transmogrifier to identify how many CPU cores there are and attempt an automatic, conservative setting.

What are the relevant tickets?

• https://mitlibraries.atlassian.net/browse/TIMX-738

Code review

• Code review best practices are documented here and you are encouraged to have a constructive dialogue with your reviewers about their preferences and expectations.

[ghukill]

This was a surprising finding for me. I had not considered the ramifications of our crosswalk data stored as XML, and using bs4 for looking up values.

By caching values as we parse and find them, we get a 2x speedup for alma transformations.

The approach may seem opaque at first, but it's pretty conceptually simple:

• create this _loc_crosswalk_code_cache cache object
• as we encounter a crosswalk type + value (e.g. "locations" and "Stacks") we check if a cached crosswalk value exists, and return if so
• if not, we use the same bs4 find(...) approach, get the value and cache it for next time

[ehanson8]

Interesting, I wasn't expecting that much of an effect either

[ghukill]

This part is critical.

Formerly, the Transformer.__iter__() method returned self, indicating that if we "iterate overe" a Transformer instance we'd just keep calling __next__(). That is purely sequential.

Now, if workers > 1, iterating over a Transformer instance will invoke self._iter_with_workers() which introduces the parallel processes.

[ghukill]

As noted above, if workers > 1, then we use this method skipping the simpler __iter__() --> __next__() approach previously used.

Now, we do work in two distinct pieces:

1- get a chunk of input source records via:

for chunk in itertools.batched(
    self.source_records, WORKER_CHUNK_SIZE, strict=False
):

2- and for each chunk, transform via module level helper function _transform_in_worker():

pending.append(
    (
        records,     <-------- chunk of records
        executor.submit(
            _transform_in_worker,  <---------------- function to transform
            [(record, next(offsets)) for record in records],
        ),
    )
)

If we look at _transform_in_worker we see that it uses _process_source_record, just like __next__() for a single worker approach:

def _transform_in_worker(chunk: list[tuple[bytes, int]]) -> list[TransformResult]:
    """Transform a chunk of (source record, run_record_offset) in a worker process."""
    if _worker_transformer is None:
        raise RuntimeError("Worker transformer was not initialized")
    return [
        _worker_transformer._process_source_record(record, run_record_offset)  # noqa: SLF001
        for record, run_record_offset in chunk
    ]

Said more generally: when we do parallel work, we can't use __next__(), because we can't pass around the Transfomer instance between processes! Therefore, both __next__() and this parallel approach both use the same method ._process_source_record() which is safe for a single or parallel context.

[jonavellecuerdo]

I always appreciate an explanation when parallel processing or multithreading is involved. Thanks for this write-up!

[ghukill]

All this code was formerly in __next__() but was moved out to be compatible with parallel worker contexts.

[Copilot]

🟡 Changes recommended

Worker-enabled incremental consumption can lose prefetched records or bypass pending results because iteration and next() do not share a persistent worker iterator.

3 open issues

• Preserve worker results across incremental iteration · New
• Test concurrent chunks and bounded-queue draining · New
• Make full-text ordering deterministic across workers · New

What changed in this PR

This PR adds optional process-based parallel transformation to Transmogrifier, building on the preceding Parquet support to make large transformation runs faster.

Changes:

• Adds --workers, ordered worker processing, and centralized record bookkeeping.
• Defers parsing until transformation and preserves original source-record bytes.
• Caches MARC crosswalk lookups, updates Parquet reading, and expands tests and documentation.

File	Description
uv.lock	Removes the jsonlines dependency.
transmogrifier/​sources/​xmltransformer.py	Reads XML records as bytes for deferred parsing.
transmogrifier/​sources/​xml/​researchdatabases.py	Adapts indexing and deletion records to byte-based input.
transmogrifier/​sources/​xml/​marc.py	Caches crosswalk lookups to reduce repeated searches.
transmogrifier/​sources/​transformer.py	Adds worker orchestration, separates bookkeeping, and updates input readers.
transmogrifier/​sources/​jsontransformer.py	Reads JSONLines records as bytes without jsonlines.
transmogrifier/​cli.py	Exposes the worker-count option.
tests/​test_cli.py	Tests worker configuration and validation.
tests/​sources/​xml/​test_researchdatabases.py	Updates expectations for byte-based reading.
tests/​sources/​xml/​test_marc.py	Tests cached crosswalk behavior.
tests/​sources/​test_transformer.py	Tests byte preservation, bookkeeping, and worker behavior.
tests/​conftest.py	Updates the transformer fixture’s reader interface.
README.md	Documents workers, performance, and source-byte preservation.
pyproject.toml	Removes the direct jsonlines dependency.

_{Review effort: Balanced}

[ehanson8]

Impressive improvement in speed, great work!

[jonavellecuerdo]

This is really exciting! I just have one small request to include more detail in a docstring.

[jonavellecuerdo]

Hmm, because this _init_worker method is specifically called as part of Transformer._iter_with_workers() method, it did give me pause to see iter(()) being passed in to the source_records parameter. From my understanding, this empty iter is not relevant / used by the _iter_with_workers() workflow but was only needed to instantiate a Transformer class. 🤔

Can you update the docstring to clarify this information?