Kiryu

Kiryu is a configuration-driven framework designed to standardize the lifecycle of data pipelines and machine learning models. Built on top of Pydantic, Polars, and Pandera, it provides a robust interface for defining, validating, and storing datasets and models.

---

✨ Key Features

• Config-Driven: Define your data and models using Pydantic schemas. Supports loading from YAML and JSON out of the box.
• ETL Lifecycle: Standardized extract(), transform(), and load() methods for repeatable data engineering.
• Model Management: Unified interface for training and serving models across Scikit-Learn, XGBoost, LightGBM, and PyTorch.
• Flexible Serialization: Support for Parquet, CSV, Arrow, ONNX, Joblib, and more.
• Hive Partitioning: Native support for partitioning timeseries data (e.g., date=2026-04-18/).
• Arrow Flight Ready: Built-in support for generating Arrow schemas and record batches for high-performance data serving.

---

🚀 Quick Start

1. Define a Dataset

Create a dataset by inheriting from Dataset and providing a configuration.

from kiryu import Dataset, DatasetConfig
import polars as pl

class UserAnalytics(Dataset):
    config = DatasetConfig(
        name="user_activity",
        version="v1",
        format_type="parquet",
        schema_fields={
            "user_id": int,
            "action": str,
            "timestamp": "datetime"
        }
    )

    def extract(self):
        # Your logic to get raw data
        return pl.DataFrame({"user_id": [1], "action": ["login"]})

    def transform(self, data):
        # Your transformation logic
        return data.with_columns(pl.lit("2026-04-18").alias("timestamp"))

# Run the pipeline
UserAnalytics().run()

2. Manage a Machine Learning Model

Standardize how you train and save models.

from kiryu import Model, ModelConfig, ModelType

class ChurnModel(Model):
    config = ModelConfig(
        name="churn_classifier",
        version="1.0.0",
        model_type=ModelType.SKLEARN,
        format_type="joblib"
    )

    def train(self):
        from sklearn.ensemble import RandomForestClassifier
        model = RandomForestClassifier()
        # ... training logic ...
        return model

    def predict(self, inputs):
        return self.model.predict(inputs)

# Train and persist
ChurnModel().train_and_save()

---

🛠 Supported Formats

Category	Formats / Frameworks
DataFrames	Pandas, Polars
Dataset Storage	Parquet, CSV, Arrow IPC
Model Types	Scikit-Learn, XGBoost, LightGBM, PyTorch
Model Storage	Pickle, Joblib, ONNX, Torch (.pt), JSON

---

📂 Project Structure

• configs.py: Pydantic models for dataset and model metadata. Includes snake_case validation for names.
• datasets.py: Base classes for standard and TimeseriesDataset (with parallel processing support).
• models.py: Abstract base class for model lifecycles.
• serializers.py: The registry system handling conversion between Python objects and disk formats.
• storages.py: Pluggable storage backends (defaulting to Local Storage).
• partitions.py: Logic for Hive-style directory structures.

---

🚦 Requirements

• Python 3.10+
• Pydantic v2
• PyYAML
• Polars (Recommended for data processing)
• Pandera (For schema validation)

---

Note: This library enforces snake_case for all dataset and model names to ensure compatibility across various storage backends and cloud providers.