TIMX 530 - prep work and s3 client#155
Closed
ghukill wants to merge 42 commits into
Closed
Conversation
Why these changes are being introduced: Bulk reading and writing from the TIMDEX dataset is a primary responsibility, but occassional random access (e.g. locating a single record row) will be helpful (e.g. looking at the original source record for a problematic record). Each TIMDEX JSON record in Opensearch will contain a "provenance" object that will include things like run_date, run_id, and now run_record_offset. This offset allows for quicker (time) and more efficient (data read) retrieval of a single record given information in the TIMDEX provenance object. How this addresses that need: Parquet files have metadata embedded that describe what values can be found in subsets of the file, but this is only helpful when the min/max values in that metadata can inform query engines if a desired record may be present. Unfortunately, the timdex_record_id is a) not lexicographically sortable (at least not easily), and b) are not ordered during write. By adding this offset, effectively an incrementing counter as records are yielded for writing, we have a value that is pre-sorted and provides nice ranges in the parquet file metadata. Query engines can utilize this to dramatically improve random access reads. By including this offset integer in the TIMDEX record "provenance" section we close the loop and provide enough information in the Opensearch record to efficiently retrieve it from the parquet dataset. Side effects of this change: * Dataset will now include a new column 'run_record_offset' Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-465
Why these changes are being introduced: During use of this library for batch reading transformed records to index into Opensearch, the application TIM threw an out-of-memory error. It was observed that, as-is, batch reading of records could hover around 1-2gb. This was surprising, as we are careful to only yield records as they are batch read from the parquet dataset. It turns out that pyarrow Dataset.to_batches() has some defaults for optimistic reading ahead to improve IO, but at the cost of memory consumption. Tuning these down resulted in much lower memory consumption, that aligns with how our current TIMDEX applications are resourced. How this addresses that need: By surfacing the Dataset.to_batches() arguments 'batch_readahead' and 'fragment_readahead' to this library's read methods, and setting conservative defaults, memory consumption is significantly lower. Side effects of this change: * Per the defaults set, slower IO for dataset reads. Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-468
Why these changes are being introduced: With the addition of two read configurations that would be passed around beteween multiple methods, the dataset module tipped to where a centralized configuration object would be helpful. Additionally, we have learned that per-operation configurations are rare, and much more likely to be set once during TIMDEXDataset init, or even as env vars for the duration of the library import. How this addresses that need: Creates a dataclass TIMDEXDatasetConfig that is passed to TIMDEXDataset on init. This class provides a typed object, with sensible defaults, that are shared throughout all read and write methods. Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-468
Why these changes are being introduced: As the parquet dataset architecture is getting promoted to production, it felt like a suitable time for a 1.0 release that would mark this library's first real usage. Relevant ticket(s): * None
Why these changes are being introduced: As of pipenv 2025.0.1 the use of `pipenv check` would throw an error, indicating that the library `safety` was not installed. It worked to run `pipenv check --auto-install` which would temporarily install `safety`, but this was not ideal for multiple reasons. First, we anticipate potentially moving away from `pipenv`. Second, it appears that `safety` is moving to a pay / subscription model. Third, it remains a little obfuscated what `pipenv check` is actually doing. As this new situation affects all builds in Github Actions CI, we need a way to scan for vulnerabilities that ideally is not a massive overhaul of our vulnerability scanning approach. How this addresses that need: `pip-audit` is a nice standalone, open-source library that performs very similar work to `safety`. This commit replaces `pipenv check` (which was `safety` under the hood) with `pip-audit`. Side effects of this change: * Builds will be successful in Github Actions Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/IN-1240
Why these changes are being introduced: One of the challenges the architecture of the TIMDEX parquet dataset presents is quick and easy metadata about ETL "runs" in the dataset. The year/month/day partitioning structure is very efficient for accessing a run if you know the date, where only a few parquet files are scanned, but it's not geared towards quickly isolating runs (parquet files) associated with a given source. Having metadata about runs provides a map to efficiently access meaningful subsets of data. One example would be fully refreshing a source in Opensearch. To do, you'd want to access all runs for a given source since, and including, the last run_type=full run. Those runs represent the current state of the source in TIMDEX. Unfortunately, this is not terribly efficient to naively perform with pyarrow or DuckDB, where potentially thousands of parquet files are touched. Similar to how Apache Iceberg (a parquet dataset architecture) works, we need some metadata about each "run" in the dataset which correlates to parquet file(s). How this addresses that need: A new class TIMDEXRunManager exists to provide this functionality. This class will produce a pandas dataframe of metadata about all runs in the dataset, including the explicit parquet filepath the run is associated with, in a highly efficient and parallelized way. The is achieved by: 1. Getting a list of all parquet files from the dataset. 2. Reading the *first* row from each file, which contains metadata about the run that produced the file. 3. Aggregating the results and grouping by "run_id". The result is a dataframe that provides a precise map of run metadata to parquet files in the dataset. With those parquet files identified, this unblocks further functionality for this library like "replaying" the runs for a given source in chronological order to refresh it in Opensearch. Side effects of this change: * None. No changes are made to pre-existing functionality, just the addition of this new information gathering class. Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-493 * https://mitlibraries.atlassian.net/browse/TIMX-494
Why these changes are being introduced: With the creation of TIMDEXRunManager we now have the ability to identify parquet files associated with current ETL runs for all or a given source. This could be used to limit the TIMDEXDataset on load to only read from those parquet files. How this addresses that need: * Updates TIMDEXDataset.load() with a new 'current_records' flag that if True will use TIMDEXRunManager to get a list of parquet files to upload the dataset paths with. Side effects of this change: * None without explicit use. Eventually, could be utilized by contexts where only parquet files associated with current ETL runs are needed. Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-494
Why these changes are being introduced: With TIMDEXDataset capable of limiting to only parquet files associated with current runs, the next logical step is providing the ability to yield only the current version of a record. This would support a "full refresh" of a TIMDEX source where an application like TIM could yield only current records for a given source and index those to Opensearch. How this addresses that need: When TIMDEXDataset is loaded with current_records=True, the private attribute TIMDEXDataset._dedupe_on_read is set to True, informing any read methods to dedupe during yielding. Because all read methods TIMDEXDataset.read_batches_iter() at the lowest level, the deduping logic is required only there. Because the ordering of the parquet files is already handled by the load method, the read methods can be confident they are always seeing the most recent version of a record first, and thus can just maintain a "seen" list as they are encountered. This keeps the deduplication effectively instant and memory safe; no large in-memory reordering or deduplication is required. Side effects of this change: * Applications like TIM now have the option of yielding only current records for a source, or all sources, supporting new functionality like fully reindexing a source in Opensearch from parquet dataset data alone. Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-494
Why these changes are being introduced: It was a bit confusing, and required unneeded logic branching, if one should use .get_current_parquet_files() or .get_current_source_parquet_files(). How this addresses that need: * .get_current_parquet_files() becomes the public interface to use, now with an optional 'source' keyword argument Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-494
Why these changes are being introduced: Formerly, an instance of TIMDEXRunManager expected a TIMDEXDataset on init, where it would utilize the pyarrow TIMDEXDataset.dataset. This results in an unneeded tightly coupling betweent these classes. How this addresses that need: * TIMDEXRunManager updated to only expect a pyarrow Dataset Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-496
Why these changes are being introduced: Unexpected behavior was possible when using load(current_records=True) and then applying additional filtering to the dataset before reading. In short, a non-current record could be yielded if filtering removed the truly current version of the record. This happened because the reverse chronological marking of "seen" records would not "see" this record and happily yield an older version. How this addresses that need: When load(current_records=True) is used, a clone of the dataset is saved to the TIMDEXDataset object before any additional filtering is applied. This dataset is just metadata, not expensive to store. Then, during any read methods, this dataset is used to provide an exhaustive and ordered list of timdex_record_ids. Even if a record has been filtered out by the read method (e.g. limiting records to only action="index"), this secondary list of timdex_record_ids is used as the authoritative list of "seen" timdex_record_ids. There is a bit of network overhead to this parallel batch reading, but fairly minimal as we are only retrieving the 'timdex_record_id'; perhaps 1-2mb of IO per millions of records. Side effects of this change: * Applications like TIM that will likely use this new functionality to yield only "current" records can do so confidently, and optionally with additional filtering, knowing they will only encounter current versions of a record from the dataset. Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-497
Why these changes are being introduced: During work on yielding only "current" records from the dataset, where ordering of the ETL runs in the dataset is critical, it was determined that more time granularity was needed for each ETL run. Currently we store the YYYY-MM-DD for each run, but if multiple runs occur on the same day, we are unable to order them more granularly. How this addresses that need: * Adds new run_timestamp to parquet dataset schema * Timestamp is minted before any runs are written, and then used for each row in the ETL run Side effects of this change: * All TIMDEX components that use this library for reading and writing will need a terraform rebuild to pick up this change. Otherwise, they need no further modification. Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-496
Why these changes are being introduced:
We have a need to perform a onetime bulk edit of the parquet dataset,
which is very similar to a SQL database migration: addition of a column
and backfilling of data.
While we may not require the strictness of SQL schema migrations, which
often can be "replayed" in order to upgrade or reverse order to downgrade,
we nonetheless would benefit from some structure and code-as-documentation
for bulk operations we perform on the actual parquet dataset.
How this addresses that need:
* Creates a new root level migrations/ folder
* Includes a README with a proposed simple structure for migrations (mostly
naming conventions)
* Includes our first migration to backfill all pre-existing parquet files
with a new 'run_timestamp' column
* The run_timestamp will be pulled from the creation date of the parquet
file in S3, which is close enough for our purposes
Side effects of this change:
* TDA becomes the location for storing code related to "migrations" for
the parquet dataset
* All pre-existing parquet files will now have a run_timestamp column
Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/TIMX-496
Why these changes are being introduced: After it was discovered that a single ETL run (run_id) has multiple, different run_timestamps in the dataset, it was clear we needed a way to pass an explicit run_timestamp for writing instead of relying on a run_timestamp minted as part of the TIMDEXDataset.write() method. How this addresses that need: DatasetRecord was given an optional run_timestamp property, that defaults to using run_date and producing a full ISO timestamp from that. Side effects of this change: * Applications like Transmogrifier can pass an explicit run_timestamp when writing to the dataset, ensuring that even multiple invocations of Transmogrifier + TIMDEXDataset.write() can share the same timestamp for an ETL run. Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-509
Why these changes are being introduced:
As outlined in more detail in TIMX-506, this class is a new, alternate approach
to providing high level metadata about the dataset to better support bulk
operations.
This class is designed to remove the need for TIMDEXRunManager, while also providing
more helpful and granular data about the dataset. Ultimately this will reduce code
complexity by relying less on implicit data contracts about the datasaet and more
on verifiable, data-in-hand about the dataset that can be used for bulk operations
(e.g. yielding current records).
How this addresses that need:
* New class TIMDEXDatasetMetadata is created
* uses DuckDB to quickly crawl the dataset, generating an in-memory or on-disk
database of record metadata
* This class is not yet wired into any TIMDEXDataset operations, this will happen
in future work
Side effects of this change:
* None
Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/TIMX-506
Why these changes are being introduced: Previously, when asked to yield only current records the class TIMDEXRunManager was used to crawl the dataset and produce a dataframe of ETL runs that informed a second iterator to consult during the yielding of records. This class was superceded by TIMDEXDatasetMetadata which can provide the same information, but also more granular record level metadata which simplifies the yielding of current records. How this addresses that need: The new TIMDEXDatasetMetadata class is used to provide a mapping of timdex_record_id to run_id that allows, during the yielding of records, to only yield a record if it's associated with the correct run_id. The net effect here is a simpler, and ultimately more accureate, way to yield current records while also utilizing a class that provides more functionality in other contexts. Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-507
Why these changes are being introduced: With the introduction of TIMDEXDatasetMetadata class, there is no functionality that TIMDEXRunManager provides that is unique. Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-507
Why these changes are being introduced: It was discovered that some parquet files, mostly those before 2025-06-27, have potentially a single, very large row group instead of maximum rows per group of 1k. This has been addressed in recent work, resulting in no files after 2025-06-28 having a single row group if total rows exceed 1k. How this addresses that need: Migration script that re-packs row groups (rewrites) when a single row group is found for a number of rows that should be split up. Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-512
Why these changes are being introduced: As we prepare for expanding the role of dataset metadata, which may be stored as a static file in S3 next to the dataset, it will be helpful to test and develop against a realistic S3 backdrop but do so locally. How this addresses that need: Configures pyarrow and DuckDB to create appropriate connections when a 'MINIO_S3_ENDPOINT_URL' env var is present. Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-515
Why these changes are being introduced: While PRAGMA configurations worked for direct parquet file reading in a local MinIO S3 instance, it did not work for DuckDB ATTACH statements. How this addresses that need: By using a DuckDB secret, setting the MinIO S3 endpoint and credentials there, other statments like ATTACH work as expected. This is overall a better approach anyways! The PRAGMA approach seemed like the only option originally, but this edge case of ATTACH forced revisiting it and this use of secrets does in fact work. Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-515
Why these changes are being introduced: With the addition of dataset/metadata assets in S3, we will need to perform actions like downloading the static DB file, uploading a new one, and deleting append deltas. How this addresses that need: Creates new utility class S3Client that performs these actions. Side effects of this change: * None Relevant ticket(s): * https://mitlibraries.atlassian.net/browse/TIMX-530
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose and background context
This PR is some small prep work and updates for the upcoming "real" changes. And, the addition of an
S3Clientclass that will be used to download, upload, and delete files in S3.How can a reviewer manually see the effects of these changes?
Nothing very exciting to see at this time!
Includes new or updated dependencies?
YES
Changes expectations for external applications?
NO
What are the relevant tickets?