TIMX 383 - pipeline tweaks for large runs

[ghukill]

Purpose and background context

This PR applies some updates to support large runs after some rough edges were discovered for a run ov ~1.6k input files, resulting in ~4.1m distinct records (~8.2m A/B versions). As we get into runs this large, some of the functional-but-not-optimized code from early development needed some adjustments.

Updates include:

• TIMX 366: removing transformed files after they are added to collated dataset (commit)
  • this introduces an optional PRESERVE_ARTIFACTS env var and config property that is used here and in later data cleanup as well; artifacts are deleted by default, but this could be helpful for debugging purposes
  • this results in a very large reduction in space, where /transformed could be 75+ gb, but the final /records dataset may only be 7-8 gb
• Improve record collation (commit)
  • add logging information during intermediate work for debugging/troubleshooting
  • be more precise with columns used for intermediate CTE tables and joins
  • introduce abdiff_record_id as an unambiguous way to identify the A/B combination of a record throughout the run
• perform diff calculations via parallel processes (commit)
• TIMX 370: create a final "records" dataset that combines the "diffs" and "metrics" datasets (commit)
  • webapp is updated to pull all information from "records" dataset instead of cherry picking from "difs" and "metrics" datasets

The net effect improves the creation and handling of datasets throughout the steps. If PRESERVE_ARTIFACTS is not set or is false, then intermediate datasets will be removed once they are no longer needed. The final, new dataset "records" contains full A/B versions for all records and the sparse matrix of modified fields, which fully supports the needs of the webapp for viewing.

The following shows an example where PRESERVE_ARTIFACTS=true:

├── 2024-11-04_19-45-24          <------------ WITH preserving artifacts
│   ├── collated
│   │   └── records-0.parquet
│   ├── diffs
│   │   └── has_diff=false
│   │       └── records-0.parquet
│   ├── metrics
│   │   └── source=libguides
│   │       └── records-0.parquet
│   ├── records
│   │   └── source=libguides
│   │       └── has_diff=false
│   │           └── records-0.parquet
│   ├── run.json
│   └── transformed
│       ├── a
│       │   ├── libguides-2024-04-03-full-transformed-records-to-index.json
│       │   └── libguides-2024-10-28-daily-transformed-records-to-index.json
│       ├── b
│       │   ├── libguides-2024-04-03-full-transformed-records-to-index.json
│       │   └── libguides-2024-10-28-daily-transformed-records-to-index.json
│       └── logs.txt

While the following shows if PRESERVE_ARTIFACTS=false or is not set (default behavior):

└── 2024-11-04_19-45-59           <------------ WITHOUT preserving artifacts
    ├── records
    │   └── source=libguides
    │       └── has_diff=false
    │           └── records-0.parquet
    └── run.json

PR Reviewing NOTE: There are a fair amount of files touched in this PR, but if reviewed by commit, each commit is fairly small and self-contained.

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

The following will allow running a job with a fair amount of input files and records:

1- Set production AWS credentials

2- Set env var:

TIMDEX_BUCKET=timdex-extract-prod-300442551476

3- Init a job, produce a CSV file of input files (results in about 630), and run a diff. NOTE: this run will take roughly 13 minutes given the large number of input files, but each fairly small. This could be slightly sped up by setting TRANSMOGRIFIER_MAX_WORKERS=9 (default is 6), but be wary this is exerting a fair amount of pressure on the system given our 10 CPU core machines.

pipenv run abdiff --verbose init-job -d output/jobs/scale -a 008e20c -b 3f77c7c

pipenv run abdiff --verbose timdex-sources-csv -o output/jobs/scale/scale.csv -s libguides,researchdatabases,gismit,gisogm,aspace

pipenv run abdiff --verbose run-diff -d output/jobs/scale -i output/jobs/scale/scale.csv

The final result should look similar, with only the "records" dataset persisting at the end:

output/jobs/scale
├── job.json
├── runs
│   └── 2024-11-05_14-50-04
│       ├── records
│       │   ├── source=aspace
│       │   │   ├── has_diff=false
│       │   │   │   └── records-0.parquet
│       │   │   └── has_diff=true                # <------ only source that had diffs for this run
│       │   │       └── records-0.parquet
│       │   ├── source=gismit
│       │   │   └── has_diff=false
│       │   │       └── records-0.parquet
│       │   ├── source=gisogm
│       │   │   └── has_diff=false
│       │   │       ├── records-0.parquet
│       │   │       └── records-1.parquet
│       │   ├── source=libguides
│       │   │   └── has_diff=false
│       │   │       └── records-0.parquet
│       │   └── source=researchdatabases
│       │       └── has_diff=false
│       │           └── records-0.parquet
│       └── run.json

NOTE: If you were to run the CLI command view-job at this point, when attempting to view sample records, you will run into an error along the lines of

duckdb.duckdb.BinderException: Binder Error: Referenced column "has_diff" not found in FROM clause!
Candidate bindings: "record_diff_matrix.related_items", "record_diff_matrix.citation"

This will be immediately resolved by TIMX-385 (which is largely complete in local spike code) which altogether removes the idea of "samples" of records by providing a filterable table of all records. This function throwing this error is removed entirely.

Other notes from test run

• The joining of A and B records during collate_ab_transforms was formerly somewhat slow, given the number of parquest files that DuckDB would scan but discard. This step is considerably faster.
• While not present in a dataset of this size, formerly the de-duping process could result in out of memory (OOM) errors. This is no longer the case given an updated SQL query for de-duping in collate_ab_transforms.
• The calculation of A/B diffs was also a slow process, given it was performed sequentially for all records. By running in true parallelism (multiple CPU cores), this is also considerbly quicker.

Includes new or updated dependencies?

NO

Changes expectations for external applications?

NO

What are the relevant tickets?

• https://mitlibraries.atlassian.net/browse/TIMX-383
• https://mitlibraries.atlassian.net/browse/TIMX-366
• https://mitlibraries.atlassian.net/browse/TIMX-370

Developer

• All new ENV is documented in README
• All new ENV has been added to staging and production environments
• All related Jira tickets are linked in commit message(s)
• Stakeholder approval has been confirmed (or is not needed)

Code Reviewer(s)

• The commit message is clear and follows our guidelines (not just this PR message)
• There are appropriate tests covering any new functionality
• The provided documentation is sufficient for understanding any new functionality introduced
• Any manual tests have been performed or provided examples verified
• New dependencies are appropriate or there were no changes

[coveralls]

Pull Request Test Coverage Report for Build 11800082877

Details

• 98 of 133 (73.68%) changed or added relevant lines in 10 files are covered.
• No unchanged relevant lines lost coverage.
• Overall coverage decreased (-2.4%) to 85.265%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
abdiff/cli.py	6	9	66.67%
abdiff/core/collate_ab_transforms.py	28	31	90.32%
abdiff/core/create_final_records.py	22	35	62.86%
abdiff/core/calc_ab_diffs.py	20	36	55.56%

Totals
Change from base Build 11728992114:	-2.4%
Covered Lines:	787
Relevant Lines:	923

---

💛 - Coveralls

[ghukill]

Noting that this COALESCE(a.action, b.action) AS action remains non-deterministic. @jonavellecuerdo pointed out that two different versions of Transmogrifier could technically result in the same file getting interpreted as "to index" or "to delete" which would vary this action value.

It is proposed that we ignore for now. It remains a very unlikely edge case, and even more so, it actually exposes something that ABDiff is not equipped to handle yet: what if a timdex_record_id changes for the same intellectual record? or in this case, the action changes?

This is getting into territory we should explore sometime -- what it means when Transmog varies something fundamental about the record, beyond just the metadata -- but I would propose as out of scope to ABDiff (at this time) merely showing us differences in metadata.

[ghukill]

Side note: I learned something I had not considered about parallel workers during this work.

I had thought that ProcessPoolExecutor(max_workers=X) was sufficient to reduce CPU and memory load, but that's only partially true. While only X workers will actively use the CPU at once, they will be scheduled all together at once, as will the arguments for each worker. If the arguments -- data passed to the worker -- is large, this can actually put a large amount of data into memory before the worker even starts "working" on it.

As such, sometimes we need to actually throttle how quickly we even schedule workers. This code ensures that the "in flight" workers is less than a threshold, and we aren't scheduling too many at once.

Consider this example. We have 100gb of data, but we'll only pass 1gb to each worker to perform calculations on, and we'll only have ProcessPoolExecutor(max_workers=10). I had thought this meant only 10 workers x 1gb = 10gb in memory at any given time. But actually, we would schedule 100 workers, each with 1gb of "arguments" (input data), and therefore 100gb could be stored in memory. Sure, only 10 workers would be "working" on the 1gb they were passed each, but the waiting workers would still have their input data already in memory. This is particularly sneaky when the input arguments are coming from an iterator; it would "drain" the iterator completely (the full 100gb) to schedule all the workers, even though only 10 at once would be "working".

What we want is actually only < 10 workers even scheduled at once, thereby slowly consuming the iterator of 100gb, and keeping memory pressure low.

This is probably poorly worded, and happy to discuss more, but worth noting. FWIW, I have confirmed that this is memory safe as-is by following the pattern of throttling how many workers we have submitted to the ProcessPoolExecutor().

[ehanson8]

Informative!

[ghukill]

Update: I've made a personal note to revisit this. I know for a fact that this works, and is memory safe at scale, so perhaps it's okay to proceed with it as-is at the moment. But I'm not 100% sure what I've outlined above is accurate; I'm personally interested in digging into this a bit more, and confirming if workers can be scheduled with sizable amounts of data even before their processes are formally begun.

[ehanson8]

Great updates that are clearly informed by the challenges encountered when scaling up the input! One question (that I assume has a good answer) so consider this an approval once @jonavellecuerdo gives the ✅

[ehanson8]

Informative!

[jonavellecuerdo]

As discussed, here are a few questions I have for you! Thank you again for providing a lot of information on the key changes / improvements introduced in this PR. Definitely learning a lot from this project.

[jonavellecuerdo]

Hmm, I think these columns also need to be wrapped in a COALESCE(). Isn't it true that if:

a record exists in version B but not in version A, then these columns would be set to NULL values

Unless something changed? 🤔

[ghukill]

For all these values -- timdex_record_id, source, run_date, and run_type -- we are gauranteed they will be the same value per this join. As such, we can remove othe unecessary COALESCE() and just take the value arbitrarily from either a or b. I missed this on our first pass on this.

It's probably inconsequential, but at scale removing 4 COALESCE() calls should be a bit quicker.

[jonavellecuerdo]

Still not sure I'm 100% following. 🤔 Going back to our previous discussion where we discussed what happens if a record is present in version A but deleted in version B, isn't the code still reproducing the same situation described here where:

As we're joining records and looping through the list of transformed files, when we get to the "*_deleted.txt" file, table a will not have any records, but table b will have records. If we're only selecting the columns from a, the the joined output will look like, no? 🤔

abdiff_record_id	timdex_record_id	source	run_date	run_type	action	record_a	record_b
asdf12345					delete

[ghukill]

@jonavellecuerdo - even if a record is "deleted" in version B, it would still have column values for timdex_record_id, source, run_date, and run_type. At least it should: these can all be known from the filename of the <source>-<date>-<run-type>-to-delete.txt file that Transmogrifier produces.

If not, then you're right; it'd be dangerous to arbitrarily only take these columns from A or B.

And I think we have proof that these will always be set from here where we produce the base of the record in the intermediate "transformed" dataset, regardless of whether it came from a "to-index" or "to-delete" file produced by Transmog, ensuring these values are always present. These values come from the filename_details, which is produced by parse_timdex_filename().

Note that we still do use COALESCE() for action here. This could also be removed, and we could artibrarily take a.action or b.action knowing they are both present (again, came from the filename). I elected to leave this, as a bit of a talisman for when we revisit that issue of them potentially being different as outlined in this comment.

[ghukill]

@jonavellecuerdo, @ehanson8 - discussions with @jonavellecuerdo revealed a couple of subtle bugs with the changes in this PR. Going to push another commit with tweaks to address, but may push this review into next week! Thanks for your patience and excellent feedback so far.

[ghukill]

@jonavellecuerdo, @ehanson8 - the PR is ready for re-review. @ehanson8, I know that you gave an implicit approval, but I've re-tagged you here for review given some of these recent updates.

Thanks to @jonavellecuerdo for pushing on a couple of key areas, specifically:

• the removal of COALESCE statements during A/B joining
• related, the correctness and reliability of DeepDiff usage given unusual A or B records

I've tried to make the commits descriptive enough to stand on their own for review:

• 1ccd2f6
• eb0401f
• 8c5fe22
• 2477924

At this time, I still have not successfully performed a "full" run. Given these updates I got very close over the weekend (let it run), but encountered what appears to be an intermittent Docker engine failure. That said, I would like to continue with this PR, as I think it represents the best form so far for this application to successfully complete a "full" run (observing that a "full" run is probably not very common, but serves as a good benchmark of performance and reliability to strive for).

[ghukill]

This ratio of allowing max open files to be about 75% of the OS's limit is somewhat arbitrary, but based on some reading. 100% continued to throw errros, as the margins are too close. 50% worked, but is a decent performance degradation (i.e. half-speed possible of concurrent writes). 75% feels like a sweet spot for the moment of safe buffer to not exceed limit, but still pretty performant.

[ehanson8]

Changes look good, one potential docstring typo. Again consider this effectively an approval once @jonavellecuerdo signs off, great work!

[jonavellecuerdo]

Changes look good to me! Thank you for providing such detailed commits. ✨ Just had one question for ya'.

[jonavellecuerdo]

@ghukill When is a file closed? I figured that when the max number of rows have been written to a file, then ds.write_dataset would close it automatically. Is that not the case? Does it wait until it hits the max_open_files limit before closing any files -- as this description seems to suggest:

If greater than 0 then this will limit the maximum number of files that can be left open. If an attempt is made to open too many files then the least recently used file will be closed. If this setting is set too low you may end up fragmenting your data into many small files

[ghukill]

The crux of the issue is partitioning. Because we write the transformed files where each filename gets a partition, it almost results in a dedicated partition (directory) for each input file, which could be thousands.

Normally, this many partitions is not ideal. And we may even want to return to this and re-evaluate. But it's helpful for joining, where we can limit by partition as a glob pattern. We could also do this with individual files.

I think the pyarrow dataset writer holds folder/files open for writing when it's partitioning. If not partitioning, I think it would happily output 2k parquet files (roughly the length of the iterator for the large job). But with partitions, I think it keeps those files open, as it may need to write to them.

I think we're close in a good architecture, but we may want to revisit this. For example, even with the changes, it still consumes a lot of disk space. I'm at 100+gb for transformed files only. Granted they will get deleted as they are joined... we could theoretically performs joins right when the A and B containers are finished and avoid having those transformed files accumlate fully just to be deleted shortly thereafter.

Which is a long way of saying: I think it's related to the partioning during writing, and I think we could/should revisit at some point.