diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml index dd1802c..8c9b12e 100644 --- a/.github/workflows/docs.yml +++ b/.github/workflows/docs.yml @@ -28,7 +28,7 @@ jobs: # Build docs - name: Build docs - run: catchsegv sphinx-build docs/source docs/build + run: sphinx-build docs/source docs/build # Add nojeckyll - name: Add nojeckyll diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 67cddcd..3dea8c6 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,13 +1,4 @@ repos: - # Commitizen - - repo: https://github.com/commitizen-tools/commitizen - rev: v4.8.0 - hooks: - - id: commitizen - stages: [commit-msg] - - id: commitizen-branch - stages: [push] - # Pre-commit hooks - repo: https://github.com/pre-commit/pre-commit-hooks rev: v5.0.0 diff --git a/docs/source/api.rst b/docs/source/api.rst index c02eff1..69f59c3 100644 --- a/docs/source/api.rst +++ b/docs/source/api.rst @@ -1,22 +1,154 @@ -API reference +API Reference ============= -.. currentmodule:: oups +This section provides detailed API documentation for the main store components. -.. autosummary:: - toplevel - sublevel - ParquetSet - ParquetSet.__setitem__ - ParquetSet.__getitem__ - oups.router.ParquetHandle.pf - oups.router.ParquetHandle.pdf - oups.router.ParquetHandle.vdf +Indexer Functions +----------------- -.. autofunction:: oups.indexer.toplevel +.. autofunction:: oups.store.toplevel -.. autofunction:: oups.indexer.sublevel +.. autofunction:: oups.store.sublevel -.. autoclass:: oups.collection.ParquetSet +.. autofunction:: oups.store.is_toplevel -.. autofunction:: oups.writer.write +Core Classes +------------ + +OrderedParquetDataset +~~~~~~~~~~~~~~~~~~~~~ + +.. autoclass:: oups.store.OrderedParquetDataset + :members: + :show-inheritance: + +Store +~~~~~ + +.. autoclass:: oups.store.Store + :members: + :show-inheritance: + +Write Operations +---------------- + +.. autofunction:: oups.store.write + +Utility Functions +----------------- + +.. autofunction:: oups.store.check_cmidx + +.. autofunction:: oups.store.conform_cmidx + +Type Definitions +---------------- + +The following are important type definitions used throughout the store module: + +**Index Types** + +Indexer classes are dataclasses decorated with ``@toplevel`` that define the schema for organizing datasets. + +**Ordered Column Types** + +The ``ordered_on`` parameter accepts: + +- ``str``: Single column name +- ``Tuple[str]``: Multi-index column name (for hierarchical columns) + +**Row Group Target Size Types** + +The ``row_group_target_size`` parameter accepts: + +- ``int``: Target number of rows per row group +- ``str``: Pandas frequency string (e.g., "1D", "1H") for time-based grouping + +**Key-Value Metadata** + +Custom metadata stored as ``Dict[str, str]`` alongside parquet files. + +Examples +-------- + +Basic Usage +~~~~~~~~~~~ + +.. code-block:: python + + from oups.store import toplevel, Store, OrderedParquetDataset + import pandas as pd + + # Define indexer schema + @toplevel + class MyIndex: + category: str + subcategory: str + + # Create store + store = Store("/path/to/data", MyIndex) + + # Create sample data + df = pd.DataFrame({ + "timestamp": pd.date_range("2023-01-01", periods=1000), + "value": range(1000) + }) + + # Access dataset and write data + key = MyIndex("stocks", "tech") + dataset = store[key] + dataset.write(df=df, ordered_on="timestamp") + +Advanced Write Options +~~~~~~~~~~~~~~~~~~~~~~ + +.. code-block:: python + + from oups.store import write + + # Time-based row groups with duplicate handling + write( + "/path/to/dataset", + ordered_on="timestamp", + df=df, + row_group_target_size="1D", # Daily row groups + duplicates_on=["timestamp", "symbol"], # Drop duplicates + max_n_off_target_rgs=2, # Coalesce small row groups + key_value_metadata={"source": "bloomberg", "version": "1.0"} + ) + +Cross-Dataset Queries +~~~~~~~~~~~~~~~~~~~~~~ + +.. code-block:: python + + # Query multiple datasets simultaneously + keys = [MyIndex("stocks", "tech"), MyIndex("stocks", "finance")] + + for intersection in store.iter_intersections( + keys, + start=pd.Timestamp("2023-01-01"), + end_excl=pd.Timestamp("2023-02-01") + ): + for key, df in intersection.items(): + print(f"Processing {key}: {len(df)} rows") + +Hierarchical Indexing +~~~~~~~~~~~~~~~~~~~~~~ + +.. code-block:: python + + from oups.store import toplevel, sublevel + + @sublevel + class DateInfo: + year: str + month: str + + @toplevel + class HierarchicalIndex: + symbol: str + date_info: DateInfo + + # This creates paths like: AAPL/2023-01/ + key = HierarchicalIndex("AAPL", DateInfo("2023", "01")) diff --git a/docs/source/index.rst b/docs/source/index.rst index b931dfe..3649970 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -6,20 +6,35 @@ oups ==== -*oups* stands for Ordered Updatable Parquet Store. +*oups* stands for **Ordered Updatable Parquet Store**. -*oups* Python library provides convenience functions and class to manage a collection of parquet datasets. This includes mostly collection indexing, and ability to update ordered datasets. +*oups* is a Python library that provides powerful tools for managing collections of ordered parquet datasets. It enables efficient storage, indexing, and querying of time-series data with validated ordering and good performance. -Index ------ +Key Features +------------ + +- **Ordered Storage**: Validates data ordering within datasets +- **Schema-based Indexing**: Hierarchical organization using dataclass schemas +- **Incremental Updates**: Efficiently merge new data with existing datasets +- **Row Group Management**: Optimizing storage layout +- **Duplicate Handling**: Configurable duplicate detection and removal +- **Lock-based Concurrency**: Safe concurrent access to datasets +- **Cross-dataset Queries**: Query multiple datasets simultaneously + +Documentation +------------- .. toctree:: - install - quickstart - purpose - indexing - parquetset - api + :maxdepth: 2 + + install + quickstart + purpose + store + api + +Indices and Tables +------------------ * :ref:`genindex` diff --git a/docs/source/indexing.rst b/docs/source/indexing.rst deleted file mode 100644 index f6b24e0..0000000 --- a/docs/source/indexing.rst +++ /dev/null @@ -1,105 +0,0 @@ -Collection indexing -=================== - -Motivation ----------- - -Datasets are gathered in a parent directory as a collection. Each of them materializes as parquet files located in a child directory whose naming is derived from a user-defined index. - -By formalizing this index through a likewise *dataclass*, index management (user scope) is dissociated from path management (*oups* scope). - -Proposal --------- - -*oups* provides 2 class decorators for defining an indexing logic. - -* ``@toplevel`` is compulsory, and defines naming logic of the first directory level, -* ``@sublevel`` is optional, and can be used as many times as number of sub-directories are required - -By splitting indexes into different directory levels, related datasets can be gathered in common directories. -A first level could for instance specify physical quantities in different places, and a second one could specify the sampling frequency of the measures. - -Each of these levels is thus specified by a class. Those corresponding to a parent directory necessarily embed as last attribute the sub-level-related class. - -Example -------- - -.. code-block:: python - - from oups import sublevel, toplevel - - @sublevel - class Sampling: - frequency: str - @toplevel - class Measure: - quantity: str - city: str - sampling: Sampling - # Define different indexes for temperature in Berlin. - berlin_1D = Measure('temperature', 'berlin', Sampling('1D')) - berlin_1W = Measure('temperature', 'berlin', Sampling('1W')) - - # Store data in a new collection - from os import path as os_path - import pandas - from oups import ParquetSet - - dirpath = os_path.expanduser('~/Documents/code/data/weather_kbase') - ps = ParquetSet(dirpath, Measure) - dummy_data_1D = pd.DataFrame( - {'timestamp':pd.date_range('2021/01/01', '2021/01/05', freq='1D'), - 'temperature':range(10,15)}) - dummy_data_1W = pd.DataFrame( - {'timestamp':pd.date_range('2021/01/01', '2021/01/14', freq='1W'), - 'temperature':range(10,12)}) - ps[berlin_1D] = dummy_data_1D - ps[berlin_1W] = dummy_data_1W - -Created folders and files are then organized as illustrated below. - -.. code-block:: - - data - |- weather_kbase - |- temperature-berlin - |- 1D - | |- _common_metadata - | |- _metadata - | |- part.0.parquet - | - |- 1W - |- _common_metadata - |- _metadata - |- part.0.parquet - -``@toplevel`` -------------- - -``@toplevel`` decorator provides attributes and functions which are used by a ``ParquetSet`` instance to - - * generate *paths* from attributes values (``__str__`` and ``to_path`` methods), - * generate class instance (``from_path`` classmethod) - -It modifies the ``__init__`` method of decorated class so that attributes values are checked at instantiation, and use of any forbidden character or combination raises related exception. - -It also calls ``@dataclass`` class decorator, with ``order`` and ``frozen`` parameters set as ``True``. This setting enables equality between class instances with same attributes values. - -Some other characteristics are: - -* ``@toplevel`` accepts an optional ``fields_sep`` parameter to define the character separating fields (by default ``-``). This separator applies to all *levels*. - -* Decorated class can have any number of attributes (also named *fields*), but only of types ``int`` or ``str``. - -* If an attribute is a ``@sublevel``-decorated class, it is necessarily positioned last. - - -``@sublevel`` -------------- - -Likewise, - -* decorated class can have any number of attributes, but only of types ``int`` or ``str``. -* if yet another deeper *sub-level* is defined (using a ``@sublevel``-decorated class), it necessarily has to be positioned as last attribute. - -``@sublevel`` is here only an alias for ``@dataclass``, with ``order`` and ``frozen`` parameters set as ``True``. diff --git a/docs/source/install.rst b/docs/source/install.rst index 27adc24..2890af9 100644 --- a/docs/source/install.rst +++ b/docs/source/install.rst @@ -1,23 +1,78 @@ -Install -======= +Installation +============ -`Install poetry `_, then *oups*, first cloning its repository. +*oups* can be installed from source using poetry or pip. The library requires Python 3.10 or higher. + +.. note:: + *oups* is currently not published to PyPI, so installation must be done from the source repository. + +Using Poetry (Recommended) +--------------------------- + +For development or if you prefer poetry: .. code-block:: bash - # Install poetry. + # Install poetry if you don't have it curl -sSL https://install.python-poetry.org | python3 - - # Clone oups repo. + + # Clone and install oups git clone https://github.com/yohplala/oups - # Install oups. cd oups poetry install - # Test. + + # Run tests poetry run pytest -Requirements will be taken care of by `poetry`. +Using Pip (Development Install) +------------------------------- + +For development with editable install: + +.. code-block:: bash + + git clone https://github.com/yohplala/oups + cd oups + pip install -e . + + # Run tests + pytest + +Dependencies +------------ + +*oups* automatically installs these required dependencies: + +* `pandas `_ (>=2.2.3) - Data manipulation +* `numpy `_ (>=2.0) - Numerical computations +* `fastparquet `_ (>=2023.10.1) - Parquet file handling +* `numba `_ (>=0.61.2) - JIT compilation for performance +* `sortedcontainers `_ (>=2.4.0) - Efficient data structures +* `flufl-lock `_ (>=8.2.0) - File locking +* `joblib `_ (>=1.3.2) - Parallel processing +* `cloudpickle `_ (>=3.1.1) - Serialization +* `arro3-core `_ (>=0.4.6) - Arrow data processing +* `arro3-io `_ (>=0.4.6) - Arrow I/O operations + +Verification +------------ + +To verify your installation: + +.. code-block:: python + + import oups + print(oups.__version__) + + # Basic functionality test + from oups.store import toplevel + + @toplevel + class TestIndex: + name: str + version: int -* `sortedcontainers `_ -* `pandas `_ -* `vaex `_ -* `fastparquet `_ + # Test creating and using the index + test_idx = TestIndex("example", 1) + print(f"Index string representation: {test_idx}") + print("Installation successful!") diff --git a/docs/source/parquetset.rst b/docs/source/parquetset.rst deleted file mode 100644 index b0245e5..0000000 --- a/docs/source/parquetset.rst +++ /dev/null @@ -1,156 +0,0 @@ -``ParquetSet`` -============== - -Purpose and creation --------------------- - -An instance of ``ParquetSet`` class gathers a collection of datasets. -``ParquetSet`` instantiation requires the definition of a *collection path* and a dataset *indexing logic*. - -* A *collection path* is directory path (existing or not) where will be (are) gathered directories for each dataset. -* An *indexing logic* is formalized by use of a ``@toplevel``-decorated class as presented in :doc:`indexing`. - -.. code-block:: python - - from os import path as os_path - from oups import ParquetSet, toplevel - - # Define an indexing logic to generate each individual dataset folder name. - @toplevel - class DatasetIndex: - country: str - city: str - - # Define a collection path. - store_path = os_path.expanduser('~/Documents/data/weather_knowledge_base') - - # Initialize a parquet dataset collection. - ps = ParquetSet(store_path, DatasetIndex) - -Usage notes ------------ - -Dataframe format -~~~~~~~~~~~~~~~~ - -* *oups* accepts `pandas `_ or `vaex `_ dataframes. -* Row index is dropped when recording. If the index of your dataframe is meaningful, make sure to reset it as a column. This only applies to *pandas* dataframes, as *vaex* ones have no row index. - -.. code-block:: python - - pandas_df = pandas_df.reset_index() - -* Column multi-index can be recorded. Here again *vaex* has no support for column multi-index. But if your *vaex* dataframe comes from a *pandas* one initially with column multi-index, you can expand it again at recording. - -.. code-block:: python - - # With 'vaex_df' created from a pandas dataframe with column multi-index. - ps[idx] = {'cmidx_expand'=True}, vaex_df - -Writing -~~~~~~~ - -* When recording data to disk, ``ParquetSet`` instance accepts a ``tuple`` which first item is then a dict defining recording setting. Parameters accepted are those of ``oups.writer.write`` function and complementary to ``dirpath`` and ``data`` (see :doc:`api` for a review). - -.. code-block:: python - - ps[idx] = {'row_group_size'=5_000_000, 'compression'='BROTLI'}, df - -* New datasets can be added to the same collection, as long as the index used is an instance from the same ``@toplevel``-decorated class as the one used at ``ParquetSet`` instantiation. - -Reading -~~~~~~~ - -* A ``ParquetSet`` instance returns a ``ParquetHandle`` which gives access to data either through 'handles' (*vaex* dataframe or *fastparquet* parquet file) or directly as a *pandas* dataframe. - - * *fastparquet* parquet file ``ps[idx].pf``, - * or *pandas* dataframe ``ps[idx].pdf``, - * or *vaex* dataframe ``ps[idx].vdf``. - -Updating -~~~~~~~~ - -If using an *index* already present in a ``ParquetSet`` instance, existing data is updated with new one. Different keywords control data updating logic. These keywords can also be reviewed in :doc:`api`, looking at ``oups.writer.write`` function signature. - -* ``ordered_on``, default ``None`` - -This keyword specifies the name of a column according which dataset is ordered (ascending order). - - * When specified, position of the new data with respect to existing data is checked. It allows data insertion. - * It also enforces *sharp* row group boundaries, meaning that a row group will necessarily starts with a new value in column specified by ``ordered_on`` at the expense of ensuring a constant row group size. When used, no newly written row group start in the middle of duplicates values. The main motivation for this feature relates to the need to include ``ordered_on`` column to identify duplicates, as discussed in next section. - -* ``duplicates_on``, default ``None`` - -This keyword specifies the names of columns to identify duplicates. If it is an empty list ``[]``, all columns are used. - -Motivation for dropping duplicates is that new values (from new data) can replace old values (in existing data). Typical use case is that of updating *OHLC* financial datasets, for which the *High*, *Low* and *Close* values of the last candle (in-progress) can change until the candle is completed. When appending newer data, values of this last candle need then to be updated. - -The implementation of this logic has been managed as an iterative process on row groups to be written, one row group per one row group (and not over the full dataset). This makes it a low memory footprint operation. This has also 2 important implications. Make sure to understand them and check if it applies correctly to your own use case. If not, a solution for you is to prepare the data the way you intend it to be before recording it anew. - - * Duplicates in existing data that is not rewritten are not dropped. - * Conversely, duplicates in existing data that is rewritten are dropped. - * Values in ``ordered_on`` column also contribute to identifying duplicates. If not already present, ``ordered_on`` column is thus forced into the list of columns defined by ``duplicates_on``. - -* ``max_nirgs``, default ``None`` - -This keyword specifies the maximum number allowed of `incomplete` row groups. An `incomplete` row group is one that does not quite reach ``max_row_group_size`` yet (some approximations of this target are managed within the code). -By using this parameter, you allow a `buffer` of trailing `incomplete` row groups. Hence, new data is not systematically merged to existing one, but only appended as new row groups. -The interest is that an `appending` operation is faster than `merging` with existing row groups, and for adding only few more rows, `merging` seems like a heavy, unjustified operation. -Setting ``max_nirgs`` triggers assessment of 2 conditions to initiate a `merge` (`coalescing` all incomplete trailing row groups to try making `complete` ones). Either one or the other has to be met to validate a `merge`. - - * ``max_nirgs`` is reached; - * The total number of rows within the `incomplete` row groups summed with the number of rows in the new data equals or exceeds `max_row_group_size`. - -Beware that if this feature is used jointly with ``duplicates_on``, and if new data overlaps with existing data, only overlapping groups are merged together. 'Full' coalescing (i.e. with all trailing incomplete row groups) is triggered only if one the abovementionned condition is met. - -.. code-block:: python - - # Reusing previous variables. - # Initiating a new dataset - ps[idx1] = df1 - # Appending the same data. - ps[idx1] = {'max_nirgs': 4}, df1 - # Reading. - ps[idx1].pdf - Out[2]: - timestamp temperature - 0 2021-01-01 10 - 1 2021-01-02 11 - 2 2021-01-03 12 - 3 2021-01-04 13 - 4 2021-01-05 14 - 5 2021-01-01 10 # new appended data - 6 2021-01-02 11 - 7 2021-01-03 12 - 8 2021-01-04 13 - 9 2021-01-05 14 - -Other "goodies" -~~~~~~~~~~~~~~~ - -.. code-block:: python - - # Reusing previous variables. - # Review store content. - ps - Out[3]: - germany-berlin - japan-tokyo - - # Get number of datasets. - len(ps) - Out[4]: 2 - - # Delete a dataset (delete data from disk). - del ps[idx1] - ps - Out[5]: japan-tokyo - - # 'Discover' an existing dataset collection. - # (initial schema definition is needed) - ps2 = ParquetSet(store_path, DatasetIndex) - ps2 - Out[6]: japan-tokyo - - # Get min and max from a column of a given dataset. - min_, max_ = ps2[idx2].min_max(col='temperature') diff --git a/docs/source/purpose.rst b/docs/source/purpose.rst index 1477cbd..dfa3f96 100644 --- a/docs/source/purpose.rst +++ b/docs/source/purpose.rst @@ -4,21 +4,104 @@ Why *oups*? Purpose ------- -Targeting the management of 'large-size' collections of ordered datasets (more specifically time series), *oups* provides convenience class and functions to ease their identification, creation, update, and loading. -These datasets may contain different data (from different channels or feeds) or be the results of different processings of the same raw data. +*oups* (Ordered Updatable Parquet Store) is designed for managing large collections of ordered datasets, particularly time-series data. It provides a comprehensive framework for efficient storage, indexing, and querying of structured data with validated ordering. -* *oups* most notably hides path management. By decorating a likewise ``dataclasss`` with ``@toplevel`` decorator, this class is turned into an index generator, with all attributes and functions so that path to related datasets can be generated. -* it also provides an *efficient* update logic suited for ordered datasets (low memory footprint). +**Key Design Goals:** + +* **Schema-driven Organization**: Use dataclass schemas to automatically organize and discover datasets +* **Ordered Storage Validation**: Verify strict ordering within datasets for optimal query performance +* **Efficient Updates**: Support incremental data updates with intelligent merging strategies +* **Memory Efficiency**: Minimize memory footprint during read/write operations +* **Concurrent Access**: Provide safe concurrent access through file-based locking +* **Flexible Querying**: Enable cross-dataset queries and range-based data retrieval + +**Core Features:** + +* **Hierarchical Indexing**: Define complex organizational schemas using ``@toplevel`` decorated dataclasses +* **Row Group Management**: Use of parquet file structure to optimize both storage and query performance +* **Duplicate Handling**: Configurable duplicate detection and removal +* **Metadata Support**: Rich metadata storage alongside datasets +* **Range Queries**: Efficient querying across time ranges and multiple datasets simultaneously + +Use Cases +--------- + +You may think of using *oups* for: + +* **Financial Time Series**: Managing market data, trading records, and risk metrics across multiple instruments +* **IoT Data Collection**: Organizing sensor data from multiple devices and locations +* **Analytics Pipelines**: Storing intermediate and final results of data processing workflows +* **Research Data**: Managing experimental datasets with complex organizational requirements Alternatives ------------ -Other libraries out there already exist to manage collections of datasets, +Several alternatives exist for managing dataset collections: + +**Arctic (MongoDB-based)** + - Provides powerful time-series storage + - Requires MongoDB infrastructure + - More complex deployment and maintenance + +**PyStore (Dask-based)** + - Supports parallelized operations + - Less flexible organizational schemas + - `Performance concerns `_ in some scenarios + +**DuckDB or DataFusion** + - Excellent query performance + - SQL-based querying + +**Direct Parquet + File Management** + - Maximum control over file structure + - Requires implementing indexing, updates, and concurrency manually + - This is how *oups* started + +*oups* Advantages +------------------ + +Compared to these alternatives, *oups* offers: + +* **Pure Python Implementation**: No external database dependencies +* **Flexible Duplicate Handling**: User-defined logic for handling duplicate rows +* **Automated Path Management**: Schema-driven directory organization +* **Incremental Updates**: Efficient merging of new data with existing datasets +* **Ordering Validation**: Built-in verification of data ordering for optimal performance +* **Simple API**: An interface not requiring SQL knowledge +* **Lock-based Concurrency**: Safe concurrent access without complex coordination + +Example Comparison +------------------- + +**Traditional Approach:** + +.. code-block:: python + + # Manual path management + path = f"/data/{symbol}/{year}/{month}/data.parquet" + + # Manual duplicate handling + existing_df = pd.read_parquet(path) + new_df = pd.concat([existing_df, new_data]) + new_df = new_df.drop_duplicates().sort_values('timestamp') + new_df.to_parquet(path) + +**With oups:** + +.. code-block:: python -* many that I have not tested, for instance `Arctic `_, -* one that I have tested, `pystore `_. Being based on Dask, it supports parallelized reading/writing out of the box. Its update logic can be reviewed in `collection.py `_. Not elaborating about its `possible performance issues `_, and only focusing on this logic applicability, current procedure implies that any duplicate rows be dropped, except last (duplicate considering all columns, but not the index, the latter being necessarily a ``DatetimeIndex`` as per *pystore* implementation). But this hard-coded logic `may not suit all dataflows `_. + @toplevel + class DataIndex: + symbol: str + year: str + month: str -In comparison, current version of *oups*, + store = Store("/data", DataIndex) + key = DataIndex("AAPL", "2023", "01") -* is not based on Dask but directly on `fastparquet `_. No parallelized reading/writing is possible. -* provides an *efficient* update function with a user-defined logic for optionally dropping duplicates. + # Automatic path management, duplicate handling, and ordering + store[key].write( + df=new_data, + ordered_on='timestamp', + duplicates_on=['timestamp', 'symbol'] + ) diff --git a/docs/source/quickstart.rst b/docs/source/quickstart.rst index 2a84c7f..ef5bde0 100644 --- a/docs/source/quickstart.rst +++ b/docs/source/quickstart.rst @@ -1,70 +1,207 @@ Quickstart ========== -ParquetSet and indexing ------------------------ +This guide will get you started with the ``oups.store`` module for managing ordered parquet datasets. -An instance of ``ParquetSet`` class gathers a collection of datasets. -``ParquetSet`` instantiation requires the definition of a *collection path* and a dataset *indexing logic*. +Basic Concepts +-------------- -**Collection path** +The store module is built around three key concepts: -It is directory path (existing or not) where will be (are) gathered directories for each dataset. +1. **Indexer**: Defines how datasets are organized using dataclass schemas +2. **OrderedParquetDataset**: Individual datasets with validated ordering +3. **Store**: Collection manager for multiple datasets -**Indexing logic** +Let's walk through a complete example. -A logic is formalized by use of a decorated class. Indices themselves are then materialized by instantiating this class, and more specifically by the instance attributes values. +Setting Up an Indexer +--------------------- -The class itself is declared just as a `dataclass `_. -``@toplevel`` is then used as a class decorator (and not ``@dataclass``). +First, define how you want to organize your datasets using a class decorated with ``@toplevel``: .. code-block:: python - from os import path as os_path - from oups import ParquetSet, toplevel + from oups.store import toplevel - # Define an indexing logic to generate each individual dataset folder name. @toplevel - class DatasetIndex: + class WeatherIndex: country: str city: str - # Define a collection path. - dirpath = os_path.expanduser('~/Documents/code/data/weather_knowledge_base') +This creates a schema where datasets will be organized in directories like ``germany-berlin/``, ``france-paris/``, etc. - # Initialize a parquet dataset collection. - ps = ParquetSet(dirpath, DatasetIndex) +Creating a Store +----------------- -Writing new data ----------------- +Create a store instance that will manage your collection of datasets: + +.. code-block:: python + + from oups.store import Store + import os + + # Define the base directory for your data collection + data_path = os.path.expanduser('~/Documents/data/weather_data') + + # Create the store + store = Store(data_path, WeatherIndex) + +Working with Datasets +---------------------- + +**Writing Data** .. code-block:: python import pandas as pd - # Index of a first dataset, for some temperature records related to Berlin. - idx1 = DatasetIndex('germany','berlin') - # Data to be recorded. - df1 = pd.DataFrame({'timestamp':pd.date_range('2021/01/01', '2021/01/05', freq='1D'), - 'temperature':range(10,15)}) - # Populate parquet collection with a first dataset. - ps[idx1] = df1 + # Create an index for Berlin weather data + berlin_key = WeatherIndex('germany', 'berlin') + + # Create sample data + df = pd.DataFrame({ + 'timestamp': pd.date_range('2023-01-01', periods=30, freq='D'), + 'temperature': range(20, 50), + 'humidity': range(30, 60) + }) + + # Get reference to the dataset (initializes the dataset if it doesn't exist) + berlin_dataset = store[berlin_key] -``weather_knowledge_base`` folder has now been created with new data. + # Write the data with timestamp ordering + berlin_dataset.write(df=df, ordered_on='timestamp') + +The directory structure will now look like: .. code-block:: - data - |- weather_knowledge_base - |- germany-berlin - |- _common_metadata - |- _metadata - |- part.0.parquet + weather_data/ + ├── germany-berlin/ + │ ├── file_0000.parquet + │ └── file_0001.parquet + ├── germany-berlin_opdmd + └── germany-berlin.lock + +**Reading Data** + +.. code-block:: python + + # Read all data back as a pandas DataFrame + result_df = berlin_dataset.to_pandas() + print(f"Dataset has {len(result_df)} rows") + + # Check dataset metadata + print(f"Ordered on: {berlin_dataset.ordered_on}") + print(f"Number of row groups: {len(berlin_dataset)}") + +Adding More Data +----------------- + +**Incremental Updates** + +.. code-block:: python + + # Add more recent data + new_df = pd.DataFrame({ + 'timestamp': pd.date_range('2023-02-01', periods=15, freq='D'), + 'temperature': range(15, 30), + 'humidity': range(40, 55) + }) + + # This will merge with existing data in the correct order + berlin_dataset.write(df=new_df, ordered_on='timestamp') + +**Adding Another City** + +.. code-block:: python -Reading existing data + # Add data for Paris + paris_key = WeatherIndex('france', 'paris') + paris_df = pd.DataFrame({ + 'timestamp': pd.date_range('2023-01-01', periods=25, freq='D'), + 'temperature': range(25, 50), + 'humidity': range(35, 60) + }) + + store[paris_key].write(df=paris_df, ordered_on='timestamp') + +Exploring Your Store --------------------- +**List All Datasets** + +.. code-block:: python + + print(f"Total datasets: {len(store)}") + + for key in store: + dataset = store[key] + print(f"{key}: {len(dataset)} row groups") + +**Query Multiple Datasets** + +.. code-block:: python + + # Query data from multiple cities for a specific time range + keys = [WeatherIndex('germany', 'berlin'), WeatherIndex('france', 'paris')] + + start_date = pd.Timestamp('2023-01-15') + end_date = pd.Timestamp('2023-01-25') + + for intersection in store.iter_intersections(keys, start=start_date, end_excl=end_date): + for key, df in intersection.items(): + print(f"Data from {key}: {len(df)} rows") + print(f"Temperature range: {df['temperature'].min()}-{df['temperature'].max()}") + +Advanced Features +----------------- + +**Time-based Row Groups** + +.. code-block:: python + + from oups.store import write + + # Organize data into daily row groups + write( + store[berlin_key], + ordered_on='timestamp', + df=df, + row_group_target_size='1D' # One row group per day + ) + +**Handling Duplicates** + +.. code-block:: python + + # Remove duplicates based on timestamp and location + write( + store[berlin_key], + ordered_on='timestamp', + df=df_with_duplicates, + duplicates_on=['timestamp'] # Drop rows with same timestamp + ) + +**Custom Metadata** + .. code-block:: python - # Read data as a pandas dataframe. - df = ps[idx1].pdf + # Add metadata to your dataset + write( + store[berlin_key], + ordered_on='timestamp', + df=df, + key_value_metadata={ + 'source': 'weather_station_001', + 'units': 'celsius', + 'version': '1.0' + } + ) + +Next Steps +---------- + +- Explore the complete :doc:`store` architecture documentation +- Learn more about indexing in :doc:`store` (Indexer section) +- Review the full :doc:`api` reference +- Understand the :doc:`purpose` and design philosophy diff --git a/docs/source/store.rst b/docs/source/store.rst new file mode 100644 index 0000000..f1a8279 --- /dev/null +++ b/docs/source/store.rst @@ -0,0 +1,267 @@ +Store Architecture +================== + +The ``oups.store`` module provides the core functionality for managing collections of ordered parquet datasets. It consists of three main components working together to provide efficient storage, indexing, and querying of time-series data. + +Overview +-------- + +The store architecture is designed around three key components: + +1. **Indexer**: Provides a schema-based indexing system for organizing datasets +2. **OrderedParquetDataset**: Manages individual parquet datasets with ordering validation +3. **Store**: Provides a collection interface for multiple indexed datasets + +Main Components +--------------- + +Indexer +~~~~~~~ + +The indexer system allows you to define hierarchical schemas for organizing your datasets using dataclasses decorated with ``@toplevel`` and optionally ``@sublevel``. This provides a structured way to organize related datasets in common directories. + +**Motivation** + +Datasets are gathered in a parent directory as a collection. Each materializes as parquet files located in a child directory whose naming is derived from a user-defined index. By formalizing this index through dataclasses, index management (user scope) is dissociated from path management (*oups* scope). + +**Decorators** + +*oups* provides 2 class decorators for defining an indexing logic: + +- ``@toplevel`` is compulsory, and defines naming logic of the first directory level +- ``@sublevel`` is optional, and can be used as many times as needed for sub-directories + +**@toplevel Decorator** + +The ``@toplevel`` decorator: + +- Generates *paths* from attribute values (``__str__`` and ``to_path`` methods) +- Generates class instances (``from_path`` classmethod) +- Validates attribute values at instantiation +- Calls ``@dataclass`` with ``order`` and ``frozen`` parameters set as ``True`` +- Accepts an optional ``fields_sep`` parameter (default ``-``) to define field separators +- Only accepts ``int`` or ``str`` attribute types +- If an attribute is a ``@sublevel``-decorated class, it must be positioned last + +**@sublevel Decorator** + +The ``@sublevel`` decorator: + +- Is an alias for ``@dataclass`` with ``order`` and ``frozen`` set as ``True`` +- Only accepts ``int`` or ``str`` attribute types +- If another deeper sub-level is defined, it must be positioned as last attribute + +**Hierarchical Example** + +.. code-block:: python + + from oups.store import toplevel, sublevel + + @sublevel + class Sampling: + frequency: str + + @toplevel + class Measure: + quantity: str + city: str + sampling: Sampling + + # Define different indexes for temperature in Berlin + berlin_1D = Measure('temperature', 'berlin', Sampling('1D')) + berlin_1W = Measure('temperature', 'berlin', Sampling('1W')) + + # When this indexer is connected to a Store, the directory structure will look like: + # temperature-berlin/ + # ├── 1D/ + # │ ├── file_0000.parquet + # │ └── file_0001.parquet + # └── 1W/ + # ├── file_0000.parquet + # └── file_0001.parquet + +**Simple Example** + +.. code-block:: python + + from oups.store import toplevel + + @toplevel + class TimeSeriesIndex: + symbol: str + date: str + + # This creates a schema where datasets are organized as: + # symbol-date/ (e.g., "AAPL-2023.01.01/") + +OrderedParquetDataset +~~~~~~~~~~~~~~~~~~~~~ + +``OrderedParquetDataset`` is the core class for managing individual parquet datasets with strict ordering validation. It provides: + +**Key Features:** + +- **Ordered Storage**: Data is stored in row groups ordered by a specified column +- **Incremental Updates**: Efficiently merge new data with existing data +- **Row Group Management**: Automatic splitting and merging of row groups +- **Metadata Tracking**: Comprehensive metadata for each row group +- **Metadata Updates**: Add, update, or remove custom key-value metadata +- **Duplicate Handling**: Configurable duplicate detection and removal +- **Write Optimization**: Configurable row group sizes and merge strategies + +**File Structure:** + +.. code-block:: + + parent_directory/ + ├── my_dataset/ # Dataset directory + │ ├── file_0000.parquet # Row group files + │ └── file_0001.parquet + ├── my_dataset_opdmd # Metadata file + └── my_dataset.lock # Lock file + +**Example:** + +.. code-block:: python + + from oups.store import OrderedParquetDataset + import pandas as pd + + # Create or load a dataset + dataset = OrderedParquetDataset("/path/to/dataset", ordered_on="timestamp") + + # Write data + df = pd.DataFrame({ + "timestamp": pd.date_range("2023-01-01", periods=1000), + "value": range(1000) + }) + dataset.write(df=df) + + # Read data back + result = dataset.to_pandas() + +Store +~~~~~ + +The ``Store`` class provides a collection interface for managing multiple ``OrderedParquetDataset`` instances organized according to an indexer schema. + +**Key Features:** + +- **Schema-based Organization**: Uses indexer schemas for dataset discovery +- **Lazy Loading**: Datasets are loaded on-demand +- **Collection Interface**: Dictionary-like access to datasets +- **Cross-dataset Operations**: Advanced querying across multiple datasets +- **Automatic Discovery**: Finds existing datasets matching the schema + +**Example:** + +.. code-block:: python + + from oups.store import Store + from oups.store import toplevel + + @toplevel + class StockIndex: + symbol: str + year: str + + # Create store + store = Store("/path/to/data", StockIndex) + + # Access datasets + aapl_2023 = store[StockIndex("AAPL", "2023")] + + # Iterate over all datasets + for key in store: + dataset = store[key] + print(f"Dataset {key} has {len(dataset)} row groups") + +Advanced Features +----------------- + +Write Method +~~~~~~~~~~~~ + +The ``write()`` function provides advanced data writing capabilities: + +**Parameters:** + +- ``row_group_target_size``: Control row group sizes (int or pandas frequency string) +- ``duplicates_on``: Specify columns for duplicate detection +- ``max_n_off_target_rgs``: Control row group coalescing behavior +- ``key_value_metadata``: Store custom metadata (supports add/update/remove operations) + +**Example:** + +.. code-block:: python + + from oups.store import write + + # Write with time-based row groups and metadata + write( + "/path/to/dataset", + ordered_on="timestamp", + df=df, + row_group_target_size="1D", # One row group per day + duplicates_on=["timestamp", "symbol"], + key_value_metadata={ + "source": "market_data", + "version": "2.1", + "processed_by": "data_pipeline" + } + ) + + # Update existing metadata (add new, update existing, remove with None) + write( + "/path/to/dataset", + ordered_on="timestamp", + df=new_df, + key_value_metadata={ + "version": "2.2", # Update existing + "last_updated": "2023-12-01", # Add new + "processed_by": None # Remove existing + } + ) + +iter_intersections +~~~~~~~~~~~~~~~~~~ + +The ``iter_intersections()`` method enables efficient querying across multiple datasets with overlapping ranges: + +**Key Features:** + +- **Range Queries**: Query specific ranges (time, numeric, etc.) across multiple datasets +- **Intersection Detection**: Automatically finds overlapping row groups +- **Memory Efficient**: Streams data without loading entire datasets +- **Synchronized Iteration**: Iterates through multiple datasets in sync + +**Example:** + +.. code-block:: python + + # Query multiple datasets for overlapping data + keys = [StockIndex("AAPL", "2023"), StockIndex("GOOGL", "2023")] + + for intersection in store.iter_intersections( + keys, + start=pd.Timestamp("2023-01-01"), + end_excl=pd.Timestamp("2023-02-01") + ): + for key, df in intersection.items(): + print(f"Data from {key}: {len(df)} rows") + +Best Practices +-------------- + +1. **Indexer Design**: Design your indexer schema to match your data access patterns +2. **Ordered Column**: Choose an appropriate column for ordering (typically timestamp) +3. **Row Group Size**: Balance between query performance and storage efficiency +4. **Duplicate Handling**: Use ``duplicates_on`` when data quality is a concern +5. **Metadata**: Use key-value metadata to store important dataset information + + +See Also +-------- + +- :doc:`api` - Complete API reference +- :doc:`quickstart` - Getting started guide diff --git a/oups/__init__.py b/oups/__init__.py index 5b39f8f..e5d98f1 100644 --- a/oups/__init__.py +++ b/oups/__init__.py @@ -5,12 +5,22 @@ @author: yoh """ +import sys + # Import version dynamically from Poetry from importlib.metadata import PackageNotFoundError from importlib.metadata import version -from .aggstream import AggStream -from .aggstream import by_x_rows + +# Conditional import for aggstream to avoid numba issues during documentation builds +if "sphinx" in sys.modules: + # During documentation builds, skip aggstream imports + AggStream = None + by_x_rows = None +else: + from .aggstream import AggStream + from .aggstream import by_x_rows + from .store import OrderedParquetDataset from .store import Store from .store import conform_cmidx diff --git a/oups/store/ordered_parquet_dataset/write/write.py b/oups/store/ordered_parquet_dataset/write/write.py index 7705361..f981667 100644 --- a/oups/store/ordered_parquet_dataset/write/write.py +++ b/oups/store/ordered_parquet_dataset/write/write.py @@ -180,6 +180,9 @@ def write( ordered_on=ordered_on, ) df_ordered_on = Series([]) if df is None else df.loc[:, ordered_on] + # Check that df_ordered_on is sorted. + if not df_ordered_on.is_monotonic_increasing: + raise ValueError("'df_ordered_on' must be sorted in ascending order.") ordered_parquet_dataset = ( # Case 'dirpath' is a path to a directory. import_module("oups.store.ordered_parquet_dataset").OrderedParquetDataset( diff --git a/pyproject.toml b/pyproject.toml index f9f10ab..3f8d5d4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,13 +1,13 @@ [tool.poetry] name = "oups" -version = "0.1.1" +version = "2025.06.4" description = "Collection of parquet datasets." authors = ["Yohplala"] license = "Apache-2.0" repository = "https://github.com/yohplala/oups" [tool.poetry.dependencies] -python = ">=3.10" +python = ">=3.10,<4.0" sortedcontainers = ">=2.4.0" pandas = ">=2.2.3" fastparquet = ">=2023.10.1" @@ -23,7 +23,9 @@ flufl-lock = ">=8.2.0" black = ">=25.1" pre-commit = ">=4.2.0" pytest = ">=8.3.5" -commitizen = ">=4.8.0" +sphinx = ">=7.2.6" +sphinx-rtd-theme = ">=3.0.0" +numpydoc = ">=1.8.0" [build-system] requires = [ @@ -34,12 +36,3 @@ build-backend = "poetry.core.masonry.api" [tool.pytest.ini_options] testpaths = ["tests"] - -# Commitizen configuration using Poetry version provider -[tool.commitizen] -name = "cz_conventional_commits" -version_provider = "poetry" -tag_format = "$version" -update_changelog_on_bump = true -annotated_tag = true -changelog_incremental = true