Framework for building modular, event-driven data pipelines
Project description
FinDrum
FinDrum is a lightweight Python framework for building and orchestrating data pipelines with extensible architecture via operators, datasources, schedulers, and triggers.
This repository (FinDrum-Platform) is the core package and is meant to be used as a library. Custom logic (pipelines and extensions) should be defined in external projects.
Installation
pip install findrum-platform
Overview
Findrum pipelines are defined in YAML files and can include:
- A sequence of operators
- A datasource (provides data from external source)
- A scheduler (to run periodically)
- An event trigger (to respond to real-time events)
Example structures:
Example with scheduler and batch datasource:
scheduler:
type: MyCustomScheduler
pipeline:
- id: batch_ingest
datasource: MyDataSource
params:
key: value
- id: transform
operator: MyOperator
depends_on: batch_ingest
params:
key: value
Example with event and pipeline
event:
type: MyTrigger
config:
key: value
pipeline:
- id: step1
operator: MyOperator
depends_on: MyTrigger
params:
key: value
- id: step2
operator: DownstreamOperator
depends_on: step1
Interfaces
Findrum provides a minimal interface for each pipeline component. These are abstract base classes that must be subclassed by your custom logic.
Operator – Core processing unit
from findrum.interfaces import Operator
class MyOperator(Operator):
def run(self, input_data):
...
Use when defining a step in a pipeline. Must implement run(input_data). We recommend that it returns a pandas.DataFrame.
DataSource – Step that starts a pipeline
from findrum.interfaces import DataSource
class MySource(DataSource):
def fetch(self, **kwargs):
...
We recommend that it returns a pandas.DataFrame. It feeds the pipeline with data.
Scheduler – Periodic trigger for pipelines
from findrum.interfaces import Scheduler
class MyScheduler(Scheduler):
def register(self, scheduler):
# e.g., add job to APScheduler instance
...
Implements logic to execute the pipeline on a time interval or schedule.
EventTrigger – React to system/file/bucket events
from findrum.interfaces import EventTrigger
class MyTrigger(EventTrigger):
def start(self):
# Starts a file watcher, webhook listener, etc.
...
Runs the pipeline when an external event occurs (e.g., new Kafka message, file in MinIO). The trigger should call self.emit(data) to push input into the pipeline.
Core Classes
You can import and use the main classes provided by Findrum:
from findrum import Platform
Platform: Main entrypoint to manage pipelines, register them, and run based on schedule or events.
CLI Usage: findrum-run
After installing findrum-platform, a CLI tool is available:
Run a pipeline immediately
findrum-run pipelines/my_pipeline.yaml
Use a custom config file for extensions
findrum-run pipelines/my_pipeline.yaml --config config/config.yaml
Enable logging (INFO level)
findrum-run pipelines/my_pipeline.yaml --verbose
Extension Discovery
Findrum requires a config.yaml file with registered class paths:
operators:
- my_project.operators.MyCustomOperator
datasources:
- my_project.datasources.MyDataSource
schedulers:
- my_project.schedulers.MyScheduler
triggers:
- my_project.triggers.MyTrigger
This lets Findrum dynamically import your components.
Minimal Example For a Non-CLI runner
from findrum import Platform
platform = Platform("config.yaml")
platform.register_pipeline("pipelines/my_pipeline.yaml")
platform.start()
You can also run your pipelines from a python file (like main.py for example) following the example above.
Clean Project Structure
A typical project using Findrum should look like:
your-project/
├── operators/
│ └── my_operator.py
├── schedulers/
│ └── my_scheduler.py
├── triggers/
│ └── my_trigger.py
├── datasources/
│ └── my_datasource.py
├── pipelines/
│ └── my_pipeline.yaml
├── config.yaml
└── main.py (optional)
Getting Started With Examples
To get started quickly, FinDrum includes runnable examples in the examples/ folder.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file findrum_platform-1.2.2.tar.gz.
File metadata
- Download URL: findrum_platform-1.2.2.tar.gz
- Upload date:
- Size: 12.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.1.0 CPython/3.12.11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
1658fbcd1d50e92476a824db98868a03e18f4585f99ed21ad4661ed9007a4bb1
|
|
| MD5 |
79ccc1cea91ca42b4632b221c1469690
|
|
| BLAKE2b-256 |
d54689c35b6c92dc7e545fb63bc6d4010d550fd3ad87d9dca97f159e7a0a1873
|
File details
Details for the file findrum_platform-1.2.2-py3-none-any.whl.
File metadata
- Download URL: findrum_platform-1.2.2-py3-none-any.whl
- Upload date:
- Size: 14.4 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.1.0 CPython/3.12.11
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
0a9814e193c1eb6703d8f4f548d433303fedf09ca1681d7beb65f316a342eb9f
|
|
| MD5 |
e502e0cd8132de879fcc0cacf2821b9a
|
|
| BLAKE2b-256 |
53783375ba3af1619032817df4fa55f098344ccd8a7b694eb40f1955ce7c50ff
|