A repository for hosting the dynamicio library, used as a wrapper for pandas i/o operations.

-- Logo illustrated by Nick Loucas

Table of Contents

• Why wrap your i/o?
  • Managing various Resources
  • Managing Various Data Types
  • Validations & Metrics Generation
  • Testing (Running Local Regression Tests)
  • So, what do we do about these?
  • The Solution
  • Main features
• Supported sources and data formats:
  • S3 I/O
  • Athena I/O
  • Coming soon
• Installation
• API Documentation
• How to use
  • Keywords
  • Let's start
    • Step 1: Resource Definitions
    • Step 2: Defining your environment variables
    • Step 3: Read in your resource definitions
    • Step 4: Loading the data resources
      • Step 4.1. SCHEMA_FROM_FILE
      • Step 4.2. Use the dynamicio cli
      • Step 4.3: Loading from S3
      • Step 4.3: Loading from Postgres
      • Step 4.3: Loading from Athena
    • Step 5: Writing out
    • Step 6: Full Code
  • Utilising asyncio
• Testing Locally
• Last notes

Table of contents generated with markdown-toc

Why wrap your i/o?

Working with pandas dataframes has opened up a new world of potential in Data Science. However, if you are using pandas to support production pipelines, whether ML or ETL, you end up having a big part of your code be concerned with I/O operations.

Managing various Resources

First, it's the various type of resources you need to interact with; object storage (S3 or GCS) databases (Athena, Big Query, Postgres), Kafka and many more. For each of these, you have dependencies on various libraries such as fsspec, gcfs, boto3, aws-wrangler, sql-alchemy, tables, kafka-python and many more.

Managing Various Data Types

Then it's the various data types you need handle, parquet, hdfs, csv, json and many others, each of which come with their own set of configuration kwargs, things like the orientation of the dataset (json) or the parquet engine you want to use behind the scenes (pyarrow or fastparquet).

Validations & Metrics Generation

Then, it's the need to validate your expectations on the datasets; things like unique or null values being allowed in a column, allowing only a specific set of categorical values, or numerical values within a specified range. And what about metrics generation? The ability to monitor data distributions and how various metrics change with every run, is a significant aspect of monitoring the quality of your solution.

Testing (Running Local Regression Tests)

Finally, what about testing your code in different environments? Take, for instance, a traditional setup where you have the following 4 environments to work against:

• Local;
• Develop;
• Staging, and;
• Production.

Configuring your code to work against the last 3 Develop, Staging and Production can easily be done through environment variables, but what about testing locally? What if you want to run your pipelines locally? Well, you can, but usually that entails a big deal of mocking calls to external services. Instead, wouldn't it be great if you could seamlessly direct your I/O operations to local sample data.

So, what do we do about these?

This proliferation of I/O operations leads to the emergence of glue code, which can be very difficult to manage. The problem is highlighted as an ML-System Anti-Pattern in Hidden Technical Debt in Machine Learning Systems

...ML researchers tend to develop general purpose solutions as self-contained packages. A wide variety of these are available as open-source packages at places like ml-oss.org, or from in-house code, proprietary packages, and cloud-based platforms.

Using generic packages often results in a glue code system design pattern, in which a massive amount of supporting code is written to get data into and out of general-purpose packages. Glue code is costly in the long term because it tends to freeze a system to the peculiarities of a specific package; testing alternatives may become prohibitively expensive. In this way, using a generic package can inhibit improvements, because it makes it harder to take advantage of domain-specific properties or to tweak the objective function to achieve a domain-specific goal. Because a mature system might end up being (at most) 5% machine learning code and (at least) 95% glue code, it may be less costly to create a clean native solution rather than re-use a generic package.

The Solution

Quoting from the same paper:

An important strategy for combating glue-code is to wrap black-box packages into common API's. This allows supporting infrastructure to be more reusable and reduces the cost of changing packages.

Dynamicio (or dynamic(i/o)) serves exactly that; it serves as a convenient wrapper around pandas I/O operations. It's a manifestation of the dependency inversion principle--a layer of indirection if you want--which keeps your code DRY and increases re-usability, effectively decoupling business logic from the I/O layer.

Main features

dynamic(i/o) supports:

• seamless transition between environments;
• abstracting away from resource and data types through resource definitions;
• honouring your expectations on data through schema definitions;
• metrics auto-generation (logging) for monitoring purposes.

Supported sources and data formats

• S3 (or local) Input & Output, for single files (type: s3) and for prefixes holding many files (type: s3_path_prefix):
  • parquet
  • h5
  • json
  • csv
• Postgres Input & Output
• Athena Input only (type: athena)
• Kafka Output

S3 I/O

S3 is accessed directly through boto3, with one client shared per process (safe across forked workers):

• Single files (type: s3): parquet is read with one GetObject and parsed in memory, and written with one PutObject (multipart above 16 MiB), using the same pandas/pyarrow options as local I/O. awswrangler is only used when you pass an option that only it supports, and for csv/json.
• Prefixes (type: s3_path_prefix): objects are listed and fetched by a pool of 32 threads, one request per object. Parquet objects are parsed in memory as they arrive, so parsing overlaps the download. no_disk_space: true is still accepted.
• No awscli or s3fs: they are no longer dependencies, so installing dynamicio no longer puts an aws command in your environment. Credentials come from the usual boto3 chain (environment, profile, IAM role).

Compared with the previous implementation, single-file reads are about 2.4x faster and writes about 1.3x faster, and prefix reads/writes are on par or better, using less CPU per call (benchmarks, design decision).

Parquet prefixes that changed schema over time

Prefixes accumulate objects over the years, so older objects may lack columns that were added to the schema later. When reading a parquet s3_path_prefix, such objects are read with the columns they do have and the missing columns are filled with nulls; a warning lists each column and how many objects lacked it:

[s3] Some objects under s3://bucket/live/state/ lack columns that are filled with nulls: {'model_action': 2} (objects lacking each column)

This tolerance is deliberate and bounded: a column that is missing from every object raises a ValueError, and a schema column typed as a non-nullable dtype (e.g. int64) still fails, because nulls cannot be represented. Single-file reads and local_batch are strict. Watch the warning if the missing data would indicate a real upstream problem rather than an old schema (design decision).

Athena I/O

type: athena resources run a SQL query on AWS Athena and return the result as a dataframe, through AWS SDK for pandas (awswrangler). Mix WithAthena into your IO class, as with the other sources. Athena is read-only: writing raises NotImplementedError.

EVENTS_ATHENA:
  CLOUD:
    type: athena
    athena:
      s3_output: s3://my-athena-results-bucket/dynamicio/   # required: where Athena writes query results
      database: analytics        # optional, default "default"
      workgroup: primary         # optional
      region_name: eu-west-1     # optional: builds a dedicated boto3 session for that region
      query: SELECT id, ts FROM events WHERE day = '2024-01-01'   # optional here, or pass `query=` when instantiating the IO class
  schema:
    file_path: "[[ RESOURCES ]]/schemas/input/events.yaml"

Notes:

• The query comes from the resource definition or from query=... at instantiation; the latter wins, as do any other options, which are passed to awswrangler.athena.read_sql_query (unknown ones are dropped). A read without a query raises ValueError.
• Credentials and region come from the usual boto3 chain and the process-wide shared session, unless region_name is set.
• ctas_approach defaults to False: the query runs as is and column names keep the case used in the SELECT. Pass ctas_approach=True to use awswrangler's CTAS path (faster for large results, but it creates a temporary Glue table and lower-cases column names).
• awswrangler returns nullable pandas dtypes (Int64, string, boolean) and Python date objects for date columns. When the resource declares a schema, dynamicio casts to the declared dtypes, so declare them.
• Breaking (9.0): the s3_staging_dir key from the PyAthena era was removed. Rename it to s3_output (same value); a definition that still has it fails to load with an error saying so.

Coming soon

• Delta Tables (pending)
• GCS (pending)
• BigQuery (pending)

Installation

>> pip install dynamicio
# or, in a uv project
>> uv add dynamicio

To contribute, see CONTRIBUTING.md (the project is managed with uv).

API Documentation

Read our docs here: https://vortechsa.github.io/dynamicio/

How to use

We will go over an end-to-end example for reading and writing a single dataset, covering:

1. all components involved and how they are configured, and;
2. how these components are pieced together

You can find this example under the demo directory fo this repo.

Keywords:

• source configs
• resource definitions
• schema definitions

Let's start

Suppose you want to ingest the foo and bar datasets, respectively from S3 and Postgres and stage them to S3 for further processing.

Assume you want to build a pipeline that looks something like the image below:

Assume the below repository structure, which implements this pipeline, for the purpose of this tutorial:

demo
├── __init__.py
├── resources
│   ├── definitions
│   │   ├── input.yaml
│   │   ├── processed.yaml
│   │   └── raw.yaml
│   └── schemas
│       ├── input
│       │   ├── bar.yaml
│       │   └── foo.yaml
│       └── processed
│           ├── final_bar.yaml
│           └── final_foo.yaml
├── src
│   ├── __init__.py
│   ├── __main__.py
│   ├── constants.py
│   ├── environment.py
│   ├── io.py
│   ├── runner_selection.py
│   └── runners
│       ├── __init__.py
│       ├── staging.py
│       └── transform.py
└── tests
    ├── __init__.py
    ├── conftest.py
    ├── constants.py
    ├── data
    │   ├── input
    │   │   ├── bar.parquet
    │   │   └── foo.csv
    │   ├── processed
    │   │   └── expected
    │   │       ├── final_bar.parquet
    │   │       └── final_foo.parquet
    │   └── raw
    │       └── expected
    │           ├── staged_bar.parquet
    │           └── staged_foo.parquet
    ├── runners
    │   ├── __init__.py
    │   ├── conftest.py
    │   ├── test_staging.py
    │   └── test_transform.py
    ├── test_pipeline.py
    └── test_runner_selection.py

Step 1: Resource Definitions

We will start with defining our input and output resources as yaml files. These need to be defined under resources/definitions:

resources
├── __init__.py
├── definitions
│   ├── input.yaml
│   ├── processed.yaml
│   └── raw.yaml
└── schemas
    ├── input
    │   ├── bar.yaml
    │   └── foo.yaml
    └── processed
        ├── final_bar.yaml
        └── final_foo.yaml

You will need to define your pipeline's resources by creating three yaml files. The first is:

• input.yaml which concerns data read by the staging task;

---
FOO:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/input/foo.csv"
      file_type: "csv"
  actual:
    type: "s3"
    s3:
      bucket: "[[ S3_YOUR_INPUT_BUCKET ]]"
      file_path: "data/foo.h5"
      file_type: "hdf"
  schema:
    file_path: "[[ RESOURCES ]]/schemas/input/foo.yaml"

BAR:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/input/bar.parquet"
      file_type: "parquet"
  actual:
    type: "postgres"
    postgres:
      db_host: "[[ DB_HOST ]]"
      db_port: "[[ DB_PORT ]]"
      db_name: "[[ DB_NAME ]]"
      db_user: "[[ DB_USER ]]"
      db_password: "[[ DB_PASS ]]"
  schema:
    file_path: "[[ RESOURCES ]]/schemas/input/bar.yaml"

• the raw.yaml, which concerns data coming out of the staging task and go into the transform task:

---
STAGED_FOO:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/raw/staged_foo.parquet"
      file_type: "parquet"
  actual:
    type: "s3"
    s3:
      bucket: "[[ S3_YOUR_OUTPUT_BUCKET ]]"
      file_path: "live/data/raw/staged_foo.parquet"
      file_type: "parquet"

STAGED_BAR:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/raw/staged_bar.parquet"
      file_type: "parquet"
  actual:
    type: "s3"
    s3:
      bucket: "[[ S3_YOUR_OUTPUT_BUCKET ]]"
      file_path: "live/data/raw/staged_bar.parquet"
      file_type: "parquet"

• and the processed.yaml, which concerns data coming out of the *transform task:

---
FINAL_FOO:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/processed/final_foo.parquet"
      file_type: "parquet"
  actual:
    type: "s3"
    s3:
      bucket: "[[ S3_YOUR_OUTPUT_BUCKET ]]"
      file_path: "live/data/processed/final_foo.parquet"
      file_type: "parquet"
  schema:
    file_path: "[[ RESOURCES ]]/schemas/processed/final_foo.yaml"

FINAL_BAR:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/processed/final_bar.parquet"
      file_type: "parquet"
    options:
      use_deprecated_int96_timestamps: true
      coerce_timestamps: "ms"
      allow_truncated_timestamps: false
      row_group_size: 1000000
  actual:
    type: "kafka"
    kafka:
      kafka_server: "[[ KAFKA_SERVER ]]"
      kafka_topic: "[[ KAFKA_TOPIC ]]"
    options:
      compression_type: "snappy"
      max_in_flight_requests_per_connection: 10
      batch_size: 262144
      request_timeout_ms: 60000 # 60s
      buffer_memory: 134217728  # 128MB
  schema:
    file_path: "[[ RESOURCES ]]/schemas/processed/final_bar.yaml"

We will hence refer to these files as "resource definitions".

The first, input.yaml defines the input sources for the staging task, handled by the respective module (runners/staging.py) while the second one, defines its output; similarly for transform. These files are parsed by dynamicio.config.IOConfig to generated configuration i/o instances referred to as "source configs" (see demo/src/__init__.py).

Notice that under every source there are three layers: sample, actual and schema. The first two point to the variants of the same dataset, depending on whether it is called from the local environment or from the cloud (we will showcase how this distinction takes place later).

The third refers your source config to a "schema definition" for your dataset (we will cover this in detail later).

Step 2: Defining your environment variables

Also notice that paths to datasets are embedded with dynamic values identified with double squared brackets, e.g. [[ S3_YOUR_OUTPUT_BUCKET ]]. These can be defined in a module in your repository.

Resource definitions (*.yaml files) work in conjunction with global and environment variables:

• environment.py

├── __init__.py
├── src
    └── environment.py
...

Let's have a look inside.

"""A module for configuring all environment variables."""
import os

ENVIRONMENT = "sample"
CLOUD_ENV = "DEV"
RESOURCES = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../resources")
TEST_RESOURCES = os.path.join(os.path.dirname(os.path.realpath(__file__)), "../tests")
S3_YOUR_INPUT_BUCKET = None
S3_YOUR_OUTPUT_BUCKET = None
KAFKA_SERVER = None
KAFKA_TOPIC = None
DB_HOST = None
DB_PORT = None
DB_NAME = None
DB_USER = None
DB_PASS = None
REFERENCE_DATA_STATE_KEY = None
LOWER_THAN_LIMIT = 1000         # We will discuss this one later in step 4. 

This module will be passed as an input parameter to instances of the dynamicio.config.IOConfig class. Let's cover some of its variables:

• "ENVIRONMENT": "sample",
  

  used to distinguish between local and cloud runs of your module. It assumes that this environment variable is defined in the cloud environment where your module is executed from.

• "TEST_RESOURCES": os.path.join(os.path.dirname(os.path.realpath(__file__)), "../tests"),
  

  It is defined in the resource definitions, e.g.:

---
FOO:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/input/foo.csv"
      file_type: "csv"
  actual:
    type: "s3"
    s3:
      bucket: "[[ S3_YOUR_INPUT_BUCKET ]]"
      file_path: "data/foo.h5"
      file_type: "hdf"
  schema:
    file_path: "[[ RESOURCES ]]/schemas/input/foo.yaml"

and therefore needs to be defined here as well.

Any other dynamic variable (identified with the doubly squared brackets) defined in the resource definitions needs to also be defined here and can be either statically or dynamically defined (i.e. hardcoded or defined as an environment value).

Step 3: Read in your resource definitions

Reading in the resources definitions can be done by means of instantiating instances of the dynamicio.config.IOConfig class (the, so called, "source configs"). This is done in:

src
├── __init__.py

which allows it to be automatically loaded on call of any module within the pipeline package.

"""Set config IOs"""
__all__ = ["input_config", "raw_config", "processed_config"]

import logging
import os

from demo.src import environment
from demo.src.environment import ENVIRONMENT, RESOURCES
from dynamicio.config import IOConfig

logging.basicConfig(level=logging.INFO)
logging.getLogger("kafka").setLevel(logging.WARNING)


input_config = IOConfig(
    path_to_source_yaml=(os.path.join(RESOURCES, "definitions/input.yaml")),
    env_identifier=ENVIRONMENT,
    dynamic_vars=environment,
)
raw_config = IOConfig(
    path_to_source_yaml=(os.path.join(RESOURCES, "definitions/raw.yaml")),
    env_identifier=ENVIRONMENT,
    dynamic_vars=environment,
)
processed_config = IOConfig(
    path_to_source_yaml=(os.path.join(RESOURCES, "definitions/processed.yaml")),
    env_identifier=ENVIRONMENT,
    dynamic_vars=environment,
)

On loading, IOConfig will load the respective configs for all resources in the form of a multi-layered dictionary, e.g., for actual:

import demo.src.environment

{
    "FOO": {
        "sample": {
            "type": "local",
            "local": {
                "file_path": f"{demo.src.environment.TEST_RESOURCES}/data/input/foo.csv",
                "file_type": "csv",
            },
        },
        "actual": {
            "type": "s3",
            "s3": {
                "bucket": f"{demo.src.environment.S3_YOUR_INPUT_BUCKET}",
                "file_path": "data/foo.h5",
                "file_type": "hdf"
            }
        },
    }
}

Then, depending on the value of the env_identifier parameter, the respective sub-dictionary is returned. For example, with:

foo_io = input_config.get(source_key="FOO")

and with env_identifier="actual", the output would be:

"type": "s3",
"s3": {
    "bucket": f"{demo.src.environment.S3_YOUR_INPUT_BUCKET}",
    "file_path": "data/foo.h5",
    "file_type": "hdf"
}

Step 4: Loading the data resources

To load a resource, you will need to generate instances of subclasses of from dynamicio import UnifiedIO class. Note that the UnifiedIO class operates as an abstract class and cannot be used for instantiating objects.

You will need to implement your own subclasses for each of the inputs you care to load. You can do this in the io.py module, under:

.
├── src
│   ├── __init__.py
│   ├── io.py

The file looks like this:

"""Responsible for configuring io operations for input data."""
# pylint: disable=too-few-public-methods
__all__ = ["InputIO", "StagedFoo", "StagedBar"]

from sqlalchemy.ext.declarative import declarative_base

from dynamicio import UnifiedIO, WithLocal, WithPostgres, WithS3File
from dynamicio.core import SCHEMA_FROM_FILE, DynamicDataIO

Base = declarative_base()


class InputIO(UnifiedIO):
    """UnifiedIO subclass for V6 data."""

    schema = SCHEMA_FROM_FILE


class StagedFoo(WithS3File, WithLocal, DynamicDataIO):
    """UnifiedIO subclass for staged foos."""

    schema = {
        "column_a": "object",
        "column_b": "object",
        "column_c": "int64",
        "column_d": "int64",
    }


class StagedBar(WithLocal, WithPostgres, DynamicDataIO):
    """UnifiedIO subclass for cargo movements volumes data."""

    schema = {
        "column_a": "object",
        "column_b": "object",
        "column_c": "int64",
        "column_d": "int64",
    }

Instances of the DynamicDataIO class can either inherit directly from UnifiedIO (e.g. InputIO inherits from UnifiedIO) or user can choose the mixins they want to use (e.g. StagedFoo inherits from WithS3File and WithLocal mixins and needs to inherit from DynamicDataIO; note that MOR kicks in to address polymorphic conflicts--i.e. order matters). Also, all instances of the DynamicDataIO must define a class schema. The schema can have the form of a dictionary, associating columns (keys) with dtypes (values) or be defined as a yaml file (see InputIO) as explained in the next section.

N.B. For convenience's sake and to reduce the need of boilerplate code, using a single class definition like InputIO is recommended (this way all your datasets can be loaded with instances of the same class). However, if you need to use different mixins for different datasets, you can do so by defining a class for each dataset (e.g. StagedFoo and StagedBar). You will definitely need to define your own classes if you want to avoid using SCHEMA_FROM_FILE (as per the below instructions) but in this case, your dataset's name will be inferred from the dataclass name you use, e.g. StagedFoo will be inferred as STAGED_FOO.

Step 4.1. SCHEMA_FROM_FILE

from dynamicio.core import SCHEMA_FROM_FILE is a unique dynamic(i/o) object used as a placeholder. It is used to indicate that a schema is provided as part of a resource definition.

For example:

---
FOO:
  sample:
  ...
  actual:
  ...
  schema:
    file_path: "[[ RESOURCES ]]/schemas/input/foo.yaml"

foo.yaml is effectively a schema definition and looks like this:

---
name: foo
columns:
  column_a:
    type: "object"
    validations:
      has_unique_values:
        apply: true
        options: {}
    metrics:
      - Counts
  column_b:
    type: "object"
    validations:
      has_no_null_values:
        apply: true
        options: {}
    metrics:
      - CountsPerLabel
  column_c:
    type: float64
    validations:
      is_greater_than:
        apply: true
        options:
          threshold: 1000
    metrics: []
  column_d:
    type: float64
    validations:
      is_lower_than:
        apply: true
        options:
          threshold: 1000
    metrics:
      - Min
      - Max
      - Mean
      - Std
      - Variance

The file is quite self-explanatory. The format is:

DataSet:

• Column
  • type
  • validations
  • metrics

For a dataset, each of the desired columns are dictated here, along with their designated dtypes. The columns are used to filter out undesired columns in an optimal manner. This means that it will happen on loading for *.csv and *.parquet files as well as when interacting with a database, but will happen post-loading in the case of *.h5 or *.json.

dtypes are then used to validate the types of the columns. If types don't match, dynamic(i/o) will attempt to cast them and will issue a WARNING. If casting does not work either, it will throw a ValueError exception.

validations and metrics are there to document the user's expectations of the quality of the dataset. They can be automatically applied on loading or on writing out.

Specifically, you can use the following validations:

• has_unique_values # no options
• has_no_null_values # no options
• has_acceptable_percentage_of_nulls
• is_in:

    validations:
      is_in:
        apply: true
        options:
          categorical_values:
            - class_a
            - class_b
            - class_c
          match_all: false # true by default, if false, then the column unique categoricals must be equal to the acceptable ones, else they must be a subset
  

• is_greater_than

    validations:
      is_greater_than:
        apply: true
        options:
          threshold: 1000
  

• is_greater_than_or_equal # same as is_greater_than
• is_lower_than # same as is_greater_than
• is_lower_than_or_equal # same as is_greater_than
• is_between # same as is_greater_than

    validations:
      is_between:
        apply: true
        options:
          lower: 0
          upper: 1000
          include_left: false
          include_right: true # true by default
  

and metrics:

• Min
• Max
• Mean
• Std
• Variance
• Counts
• UniqueCounts
• CountsPerLabel

imposed as per below:

  column_c:
    type: float64
    validations: {}
    metrics:
      - Min
      - Max
      - Mean
      - Std
      - ...

Note that you can also use dynamic fields to define validations, e.g. see LOWER_THAN_LIMIT in the file below:

---
name: bar
columns:
  column_a:
    type: "object"
    validations:
      has_unique_values:
        apply: true
        options: {}
    metrics:
      - Counts
  column_b:
    type: "object"
    validations:
      has_no_null_values:
        apply: true
        options: {}
    metrics:
      - CountsPerLabel
  column_c:
    type: float64
    validations:
      is_greater_than:
        apply: true
        options:
          threshold: 1000
    metrics: []
  column_d:
    type: float64
    validations:
      is_lower_than:
        apply: true
        options:
          threshold: "[[ LOWER_THAN_LIMIT ]]"
    metrics:
      - Min
      - Max
      - Mean
      - Std
      - Variance

Similar to resource definitions, this value needs to be defined in environment.py

Step 4.2. Use the dynamicio cli

The dynamicio cli can be used to automatically generate schema definitions for you, provided either a path to a dataset (json, parquet, hdf, csv) or to a directory. Here is how you can use it:

usage: dynamicio [-h] (--batch | --single) -p PATH -o OUTPUT

Generate dataset schemas

optional arguments:
  -h, --help            show this help message and exit
  --batch               used to generate multiple schemas provided a datasets directory.
  --single              used to generate a schema provided a single dataset.
  -p PATH, --path PATH  the path to the dataset/datasets-directory.
  -o OUTPUT, --output OUTPUT
                        the path to the schemas output directory.

The generated schema definitions will not have any validations or metrics automatically selected for you.

Step 4.3: Loading from S3

To then load from S3 you simply do:

    foo_df = InputIO(source_config=input_config.get(source_key="FOO"), apply_schema_validations=True, log_schema_metrics=True).read()

which will load the foo.csv file as a dataframe.

Step 4.3: Loading from Postgres

Likewise to S3 resources, postgres resources need the same number of options to be defined for their loading.

Implicitly, dynamicio is able to infer data model from the schema yml files of the source key provided rather than requiring that the schema is explicitly defined. This data model defines the table, the columns and their respective SQL types.

To, then, load from postgres you simply do:

    bar_df = InputIOsource_config=input_config.get(source_key="BAR"), apply_schema_validations=True, log_schema_metrics=True).read()

which will load the cargo the movements table as a dataframe.

Step 4.3: Loading from Athena

athena resources need an s3_output location for query results and a query, either in the resource definition or passed when instantiating the IO class (which takes precedence):

    events_df = InputIO(source_config=input_config.get(source_key="EVENTS_ATHENA"), query="SELECT id, ts FROM events LIMIT 10").read()

where InputIO mixes in WithAthena. See Athena I/O for the available options.

Step 5: Writing out

Sinking data is done in a very similar way. You need to:

1. Define your output resource definitions, in our case in raw.yaml

---
STAGED_FOO:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/raw/staged_foo.parquet"
      file_type: "parquet"
  actual:
    type: "s3"
    s3:
      bucket: "[[ S3_YOUR_OUTPUT_BUCKET ]]"
      file_path: "live/data/raw/staged_foo.parquet"
      file_type: "parquet"

STAGED_BAR:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/raw/staged_bar.parquet"
      file_type: "parquet"
  actual:
    type: "s3"
    s3:
      bucket: "[[ S3_YOUR_OUTPUT_BUCKET ]]"
      file_path: "live/data/raw/staged_bar.parquet"
      file_type: "parquet"

2. You need to define the respective dynamic values found in your resource definitions in your src/environment.py
3. You need to create an instance of the IOConfig class for the raw.yaml in the __init__.py file (we already did this).
4. Define the additional DynamicDataIO subclasses in the src/io.py module, dictating through the schema the list of columns, and their types (also used for schema validation).
5. Finally, instantiate instances of those subclasses and call the .write() method, passing in the dataframe you want to write out, e.g.

demo/src/runners/staging.py:

    ...
    StagedFoo(source_config=raw_config.get(source_key="STAGED_FOO"), **constants.TO_PARQUET_KWARGS).write(foo_df)
    StagedBar(source_config=raw_config.get(source_key="STAGED_BAR")).write(bar_df)

Notice that you can pass all pandas options to write out, when for instance you are writing out parquet.

demo/src/constants.py:

# Parquet
TO_PARQUET_KWARGS = {
    "use_deprecated_int96_timestamps": False,
    "coerce_timestamps": "ms",
    "allow_truncated_timestamps": True,
}

Of, course this is not a problem as parquet is the format used by both resources in either environment. This not always the case however. See in demo/resources/definitions/processed.yaml:

---
...

FINAL_BAR:
  sample:
    type: "local"
    local:
      file_path: "[[ TEST_RESOURCES ]]/data/processed/final_bar.parquet"
      file_type: "parquet"
    options:                                      <---- Options for Local writing as parquet
      use_deprecated_int96_timestamps: true
      coerce_timestamps: "ms"
      allow_truncated_timestamps: false
      row_group_size: 1000000
  actual:
    type: "kafka"
    kafka:
      kafka_server: "[[ KAFKA_SERVER ]]"
      kafka_topic: "[[ KAFKA_TOPIC ]]"
    options:                                      <---- Options for writting to a Kafka Topic
      compression_type: "snappy"
      max_in_flight_requests_per_connection: 10
      batch_size: 262144
      request_timeout_ms: 60000 # 60s
      buffer_memory: 134217728  # 128MB
  schema:
    file_path: "[[ RESOURCES ]]/schemas/processed/final_bar.yaml"

Here, we have a case where different options need to be used for each environment as it deals with a different source. This is gracefully managed through resource definitions passing these arguments in the options key per environment.

Step 6: Full Code

The full code for the loading module in our example would live under:

├── __init__.py
├── src
...
│   ├── runners
│   │   └── staging.py

and looks like:

"""Add module docstring...."""
import logging

from demo.src import constants, input_config, raw_config
from demo.src.io import InputIO, StagedBar, StagedFoo

logger = logging.getLogger(__name__)


def main() -> None:
    """The entry point for the Airflow Staging task.

    Returns:
        Void function.
    """
    # LOAD DATA
    logger.info("Loading data from live sources...")

    bar_df = InputIO(source_config=input_config.get(source_key="BAR"), apply_schema_validations=True, log_schema_metrics=True).read()
    foo_df = InputIO(source_config=input_config.get(source_key="FOO"), apply_schema_validations=True, log_schema_metrics=True).read()

    logger.info("Data successfully loaded from live sources...")

    # TRANSFORM  DATA
    logger.info("Apply transformations...")

    # TODO: Apply your transformations

    logger.info("Transformations applied successfully...")

    # SINK DATA
    logger.info("Begin sinking data to staging area:")
    StagedFoo(source_config=raw_config.get(source_key="STAGED_FOO"), **constants.TO_PARQUET_KWARGS).write(foo_df)
    StagedBar(source_config=raw_config.get(source_key="STAGED_BAR")).write(bar_df)
    logger.info("Data staging is complete...")

Utilising asyncio

Dynamic(i/o) supports use of asyncio to speed up I/O bound operations through leveraging multithreading.

An example can be found in the second of the two demo tasks, namely, the transform.py task.

"""Add module docstring...."""
import asyncio
import logging

import demo.src.environment
from demo.src import processed_config, raw_config
from demo.src.io import InputIO, StagedBar, StagedFoo

logger = logging.getLogger(__name__)


async def main() -> None:
    """The entry point for the Airflow Staging task.

    Returns:
        Void function.
    """
    # LOAD DATA
    logger.info("Loading data from live sources...")

    [bar_df, foo_df] = await asyncio.gather(
        StagedBar(source_config=raw_config.get(source_key="STAGED_BAR")).async_read(),
        StagedFoo(source_config=raw_config.get(source_key="STAGED_FOO")).async_read()
    )

    logger.info("Data successfully loaded from live sources...")

    # TRANSFORM  DATA
    logger.info("Apply transformations...")

    # TODO: Apply your transformations

    logger.info("Transformations applied successfully...")

    # SINK DATA
    logger.info(f"Begin sinking data to staging area: S3:{demo.src.environment.S3_YOUR_OUTPUT_BUCKET}:live/data/raw")
    await asyncio.gather(
        InputIO(source_config=processed_config.get(source_key="FINAL_FOO"), apply_schema_validations=True, log_schema_metrics=True).async_write(foo_df),
        InputIO(source_config=processed_config.get(source_key="FINAL_BAR"), apply_schema_validations=True, log_schema_metrics=True).async_write(bar_df),
    )
    logger.info("Data staging is complete...")

In short, you simply need to utilise the async_read() or the async_write() methods instead, plus await and gather your calls.

Testing Locally

After following the above documentation, at this point it should be clear that dynamic(i/o) is optimised for enabling seamless local testing for your pipelines.

Simply by configuring your ENVIRONMENT's default value to sample and provided that you have the required tests data sources in the necessary directories, it becomes very simple to test your pipelines end-to-end in seconds, eliminating the need to deploy your dags and wait for their tasks to be provided access to processing resources.

All you need to do is mimic the order of execution of your tasks, running them in procedural order.

In the case of our example, you would have to:

1. Add the necessary data under tests/data:

└── tests
    ├── __init__.py
    ├── conftest.py
    ├── constants.py
    ├── data
    │   ├── input
    │   │   ├── bar.parquet
    │   │   └── foo.csv
    │   ├── processed
    │   │   └── expected
    │   │       ├── final_bar.parquet
    │   │       └── final_foo.parquet
    │   └── raw
    │       └── expected
    │           ├── staged_bar.parquet
    │           └── staged_foo.parquet
    ├── runners
    │   ├── __init__.py
    │   ├── conftest.py
    │   ├── test_staging.py
    │   └── test_transform.py
    ├── test_pipeline.py
    └── test_runner_selection.py

2. Implement an end-to-end, black-box style test that simply generates the expected data output given a specific input (deleting the output after the assertion)

An example end-to-end test in this case, for a single airflow task would look like:

"""An example pipeline to showcase how dynamicio can bt used for setting up a local e2e testing!"""
# pylint: disable=missing-module-docstring, missing-class-docstring, missing-function-docstring, unused-argument, too-few-public-methods
# noqa
import os

import pandas as pd
import pytest

from demo.src import processed_config, raw_config
from demo.src.runners import staging, transform


class TestPipeline:
    """Example e2e test."""

    @pytest.mark.end_to_end
    def test_dag_with_mock_sample_input_data(
        self,
        expected_staged_foo_df,
        expected_staged_bar_df,
        expected_final_foo_df,
        expected_final_bar_df,
    ):
        """Showcases how you can leverage dynamicio to read local data for fast feedback when you want to run your pipelines locally."""
        # Given
        # The src/resources/input.yaml

        # When
        staging.main()
        transform.main()

        # Then
        try:
            assert expected_staged_foo_df.equals(pd.read_parquet(raw_config.get(source_key="STAGED_FOO")["local"]["file_path"]))
            assert expected_staged_bar_df.equals(pd.read_parquet(raw_config.get(source_key="STAGED_BAR")["local"]["file_path"]))
            assert expected_final_foo_df.equals(pd.read_parquet(processed_config.get(source_key="FINAL_FOO")["local"]["file_path"]))
            assert expected_final_bar_df.equals(pd.read_parquet(processed_config.get(source_key="FINAL_BAR")["local"]["file_path"]))
        finally:
            os.remove(raw_config.get(source_key="STAGED_FOO")["local"]["file_path"])
            os.remove(raw_config.get(source_key="STAGED_BAR")["local"]["file_path"])
            os.remove(processed_config.get(source_key="FINAL_FOO")["local"]["file_path"])
            os.remove(processed_config.get(source_key="FINAL_BAR")["local"]["file_path"])

Last notes

Hope this was helpful.

Please do reach out with comments and your views about how the library or the docs can be improved, and by all means, come along and contribute to our project!

Changelog

v9.2.2

• fix: an IO object can be read or written repeatedly. A read or write removed values from self.options, so a second call on the same object failed or read the wrong data. The second call of a path with template fields (for example {runner_id}) raised ValueError: Expected values for all fields. Three cases gave no error. The S3 path-prefix read removed no_disk_space. The Athena read removed query, and the Postgres read removed sql_query, so the second read ran the default query and returned wrong data. Each call now works on a copy of the options, and self.options does not change. The two private methods WithS3._read_parquet_object and WithS3._concat_parquet_objects are removed. The module-level functions with the same names are unchanged

v9.2.1

• fix: transient S3 connection errors no longer fail a read or write. The shared boto3 client now uses the standard retry mode with 10 attempts, connect_timeout=10 and read_timeout=60. The in-memory and file transfers (put_object, upload_fileobj, get_object, download_fileobj) are repeated, up to 5 attempts with backoff, on ConnectionClosedError, ConnectTimeoutError, EndpointConnectionError, ReadTimeoutError, IncompleteReadError, ResponseStreamingError and the RetriesExceededError that the transfer manager raises (s3transfer's and boto3's) when a download above 16 MiB gave up on a broken part. botocore does not retry these once the body is being sent or read. The buffer is rewound before each attempt (bytes written are identical), each retry is logged at WARNING with the bucket and key, and the shared client is rebuilt so the next attempt opens fresh connections. Other errors (for example ClientError) are raised as before, and nothing changes when no error occurs

v9.2.0

• feat: a single-file S3 parquet read with filters reads only the row groups whose footer statistics can match, over ranged GetObject requests through pyarrow.fs.S3FileSystem (one for each process and bucket). The rows, index and dtypes are the same as before. Reads without filters or with filters=[], fastparquet reads, reads with a non-pyarrow filesystem, prefix reads and reads with wrangler-only kwargs do not change. On the ranged read, a missing key or a permission problem raises pyarrow FileNotFoundError or OSError in place of botocore ClientError, and a custom S3 endpoint of boto3 (AWS_ENDPOINT_URL) does not apply. See ADR 0004
• feat: new postgres write option create_table (default False). When True, the writer creates a missing table from the schema columns, with no primary key and with unbounded string columns, in the same transaction as the write. The postgres writer now lists its options (truncate_and_append, create_table), so it drops other options with a warning, as the other writers do. A create_table value that is not a bool (for example the YAML string "false") raises TypeError
• fix: the cache of pyarrow S3 file systems is cleared at exit, before the pyarrow S3 finalizer runs. With pyarrow 24 and later, a process that read with filters on a pod with web identity credentials did not exit (https://github.com/apache/arrow/issues/50188)
• chore: dev dependency moto[server]>=5 for the tests of the ranged read

v9.1.0

• chore(deps): dependency audit, ranges widened to what the code needs and CI proves. Removed packages the code never imports (fsspec, libcst, logzero, no_implicit_optional); dropped exact pins (confluent-kafka>=2.4.0,<3, psycopg2-binary>=2.9.3,<3, python-json-logger>=2.0.1, simplejson>=3.17.2); lifted the numpy<2 ceiling (numpy 2.x passes the suite) and the python<3.13 ceiling (3.13 and 3.14 pass; added to CI and classifiers). Raised floors so every allowed combination works with numpy 2 as well as numpy 1: pandas>=2.2.2 (JSON writes with index=False for records/values, relied on since 8.1.1, need pandas 2.1, and pandas before 2.2.2 breaks with numpy 2), pyarrow>=16.0.0, fastparquet>=2024.5.0, awswrangler>=3.9.0 (older ones cannot import against numpy 2 or pyarrow 16), PyYAML>=6.0 (5.4.1 does not build with current toolchains). New CI job dependency-ranges tests the minimums (Python 3.10), the minimums with numpy 2 (Python 3.11) and the newest releases (Python 3.14); publishing waits for it
• fix: python-json-logger 3.1+ import path (pythonjsonlogger.json) with fallback to the old module, avoiding a DeprecationWarning on import

v9.0.0

Breaking: Athena resources must rename s3_staging_dir to s3_output (see the first Athena entry below).

• docs: add Backstage catalog-info.yaml to enable TechDocs
• docs: README documents Athena I/O (resource options, ctas_approach, dtypes, s3_staging_dir migration); ADR 0003
• chore: migrate project and dependency management from Poetry to uv (PEP 621 pyproject.toml, uv.lock, hatchling + hatch-vcs build backend). The version is still derived from the git tag and the published wheel is unchanged apart from standard metadata. Contributors now use make dev-env-setup / uv sync; see CONTRIBUTING.md and AGENTS.md
• feat(RND-13624)!: WithAthena reads through AWS Data Wrangler (wr.athena.read_sql_query) instead of PyAthena, and the pyathena dependency is dropped. Breaking (resource definitions): the s3_staging_dir key of an athena: section is removed. Loading such a definition now fails with an error saying to rename it to s3_output (same value); nothing else changes in the YAML. region_name and query stay optional, and database (default default, Athena's own default) and workgroup are new optional keys. Reads use the shared per-process boto3 session unless region_name is set. Extra options are filtered to what read_sql_query accepts (e.g. ctas_approach, unload_approach, athena_cache_settings); chunksize is rejected with an error, because a chunked read returns an iterator rather than a DataFrame
• ctas_approach defaults to False (as PyAthena, which read the query's result file): column names keep the case used in the query (CTAS lower-cases them) and no temporary Glue table is created. Pass ctas_approach: true in options to opt in for large results
• note: results now come back with pandas nullable dtypes (Int64, string, boolean) and date as Python dates, as returned by awswrangler; declare the dtypes in the resource schema to cast them on read, as for any other source

v8.1.1

• fix: S3 JSON reads and writes use pandas' defaults again: lines=False (a JSON array for orient="records", as written by 7.x and earlier) unless the caller passes lines. 8.1.0 defaulted to JSON Lines for orient="records", which could not read existing files and wrote files that 7.x readers cannot parse. Set lines: true in the resource options to opt in to JSON Lines
• fix: S3 JSON writes default index per orient as pandas does (True for index/columns/split/table, False for records/values) instead of always False, which made orient="index" and orient="columns" writes fail or drop the index

v8.1.0-rc.11

• fix: an s3_file parquet source whose file_path ends in / (a directory of parquet objects, e.g. a downstream pipeline's outcome dataset) is read as a dataset again: all objects under it are concatenated in listing order, with the same missing-column tolerance as s3_path_prefix. Since rc.7 it failed with NoSuchKey because the single-file reader issued a GetObject on the .../ key

v8.1.0-rc.10

• fix: S3 CSV writes accept a caller-supplied index option (TypeError: to_csv() got multiple values for keyword argument 'index' since rc.1, hit by a downstream pipeline's IO(..., index=False)), and default to index=True like DataFrame.to_csv and the s3fs writer on master, instead of the hard-coded index=False of the wrangler writer
• fix: the awswrangler parquet fallback tolerates a caller-supplied dataset option instead of raising a duplicate-keyword TypeError

v8.1.0-rc.9

• perf: parquet no_disk_space prefix reads use the same in-memory thread pool as the default path (~4x faster on a 3.7M-row prefix, 199s to 46s, lower peak memory); rows, order and dtypes unchanged
• fix: parquet prefix reads return rows in S3 listing order and null-filled object columns hold None, as master's no_disk_space dataset read did (see rc.8 notes)

v8.1.0-rc.8

• perf: parquet WithS3PathPrefix reads fetch and parse each object in memory on the download threads (no temp dir, parsing overlaps the network); ~15% faster on ~1000 files over WAN, same rows. Files are now concatenated in listing order instead of arbitrary os.listdir order
• fix: WithS3PathPrefix parquet reads no longer fail with ArrowInvalid: No match for FieldRef.Name(<col>) when some objects under the prefix predate a schema column. Such objects get nulls for that column (as master's no_disk_space dataset read did when the first listed file had the column), a warning lists the columns and how many objects lacked them, and a column missing from every object still raises ValueError
• fix: parquet no_disk_space prefix reads return rows in S3 listing order again, as master did (the branch had reversed it); null-filled object columns hold None like master's dataset read, not NaN
• fix: WithPostgres logged Started downloading table twice per read (the query itself ran once); now logged once

v8.1.0-rc.7

(rc.5 and rc.6 were tagged but never published: black/pylint/flake8 style checks failed. Contents are identical apart from formatting.)

• perf: reuse one boto3 session/client per process (fork-safe) instead of building a new one per awswrangler call
• perf: single-file S3 parquet reads/writes use one GetObject/PutObject over the shared client and parse in memory with the same pandas/pyarrow options as local I/O; awswrangler is only used when a wrangler-only option is passed (fixes the ~2.6x read slowdown seen in rc.1-rc.4)
• feat!: remove awscli and s3fs dependencies; WithS3PathPrefix now syncs with a boto3 thread pool (one request per file, concurrency 32) instead of the in-process aws s3 sync. No aws console script is installed into consumer venvs any more
• fix: S3 HDF writer honours pickle_protocol
• fix: S3 JSON reader supports single_record

v8.1.0-rc.4

• fix: stop forcing convert_dates=False on local JSON reads, letting pandas' own date auto-detection run unless the caller opts out explicitly (fixes datetime dtype regression for consumers relying on auto-parsed dates, e.g. reference-data JSON)

v8.1.0-rc.3

• fix: honor caller-supplied orient/lines for S3 and local JSON reads/writes instead of hardcoding orient="records" (fixes regression for consumers using orient="index", e.g. reference-data JSON)
• fix: re-apply popped orient/lines options before df.to_json in the local JSON writer (previously silently discarded)

v8.1.0-rc.2

• fix: stop forcing records-only orient on local JSON reads

v8.1.0-rc.1

• feat(RND-13624): Migrate WithS3File to AWS Data Wrangler
• fix: address Copilot review findings on WithS3File wrangler migration
• fix: silence pre-existing pylint invalid-name warnings to unblock style-checks
• fix: satisfy black formatting and 90% coverage gate in CI
• fix: regenerate poetry.lock to match pyproject.toml
• docs: add awswrangler vs AWS CLI performance report, PR template, and skills