Metadata-Version: 2.4
Name: pymapreduce-core
Version: 0.1.9
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Requires-Dist: cloudpickle>=3.0.0
Summary: A high-performance MapReduce framework in Rust with Python async/await bindings
Author: MapReduce Team
License: MIT
Requires-Python: >=3.7
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM

# 🐍 PyMapReduce

<p align="center">
  <a href="https://pypi.org/project/pymapreduce/"><img src="https://img.shields.io/badge/pypi-v0.1.3-blue.svg?style=flat-square" alt="PyPI version"/></a>
  <a href="https://www.python.org/downloads/"><img src="https://img.shields.io/badge/python-3.8%2B-brightgreen.svg?style=flat-square" alt="Python 3.8+"/></a>
  <a href="https://github.com/PyO3/pyo3"><img src="https://img.shields.io/badge/PyO3-Rust_Bindings-orange.svg?style=flat-square" alt="PyO3"/></a>
  <a href="https://github.com/tokio-rs/tokio"><img src="https://img.shields.io/badge/AsyncIO-Tokio_Powered-blueviolet.svg?style=flat-square" alt="AsyncIO"/></a>
</p>

`pymapreduce` is the official Python SDK for the **MapReduce Distributed Framework**, a high-performance distributed computing engine written in Rust. Inspired by **Ray**, `pymapreduce` brings the speed, memory safety, and concurrency of Rust to Python developers through a clean, intuitive, and asynchronous API.

---

## 🚀 Key Capabilities

- **Ray-Like Developer Experience**: Use `@pymapreduce.remote` to turn ordinary Python functions and classes into distributed tasks and stateful actors.
- **Stateful Actor Support**: Maintain class instance state (e.g., ML weights, database connections, caching layers) in-memory across distributed method invocations.
- **Zero-Copy Tiered Object Store**: Pass large datasets (`ObjectRef`) between tasks with automatic peer-to-peer fetching without routing bulky payloads through the driver.
- **True Multi-Core Parallelism (GIL-Free)**: Each worker runs in an isolated Python process via high-speed IPC (stdin/stdout binary framing), bypassing Python's Global Interpreter Lock (GIL).
- **High-Performance `runtime_env`**: Automatically packages project source code with `.gitignore` and default excludes, extracts once into shared memory (`/dev/shm`), and provides 3-tier automatic lifecycle cleanup.
- **$O(1)$ Load-Balancing & Adaptive Scheduler**: Dynamic task routing preventing deadlocks and maximizing CPU utilization across multiple physical nodes.
- **Embedded & Distributed Modes**: Seamlessly transition from a local zero-setup embedded cluster (`pymapreduce.init()`) to a multi-node production deployment.

---

## 📦 Installation & Build

### Prerequisites
1. [Rust Toolchain (Cargo)](https://rustup.rs/) (version 1.75+)
2. Python 3.8+

### Building from Source (Development Mode)
```bash
# 1. Create and activate a virtual environment
python3 -m venv .venv
source .venv/bin/activate

# 2. Install Maturin build tool
pip install maturin cloudpickle

# 3. Compile and install in editable development mode
maturin develop --release
```

### Building Distribution Wheels
```bash
maturin build --release
pip install target/wheels/pymapreduce_core-0.1.3-*.whl
```

---

## 💻 Usage Guide

### 1. Cluster Operation

#### Option A: Running with Python CLI (`mapreduce`)
```bash
# Start HeadNode (Central scheduler & GCS on port 7777)
mapreduce start --head --port 7777 --scheduler Adaptive

# Start WorkerNode on compute machines
mapreduce start --worker --head-addr 127.0.0.1:7777 --cpus 4
```

#### Option B: Embedded Single-Node Mode (Notebooks & Local Dev)
You can spawn a local cluster programmatically in your application with zero setup:
```python
import asyncio
import pymapreduce

async def main():
    # Automatically starts local HeadNode
    driver = await pymapreduce.DriverWrapper.init()

    # Launch background worker with 4 cores
    asyncio.ensure_future(pymapreduce.start_worker("127.0.0.1:7777", 4))
    await asyncio.sleep(2) # Allow worker to register
```

---

### 2. Programming Paradigms

#### A. Stateless Distributed Tasks (Functions)
```python
import asyncio
import pymapreduce

# Decorate function to run remotely on worker nodes
@pymapreduce.remote
def square(x: int) -> int:
    return x * x

async def main():
    # Connect to the cluster
    driver = await pymapreduce.DriverWrapper.init("127.0.0.1:7777")

    # 1. Execute a single task
    res = await square.remote(10)
    print("Single Task Result:", res) # 100

    # 2. Execute parallel tasks across the cluster
    tasks = [square.remote(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    print("Parallel Results:", results) # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

if __name__ == "__main__":
    asyncio.run(main())
```

---

#### B. Stateful Distributed Actors (Classes)
Actors maintain persistent state in worker memory across multiple remote method calls:
```python
import asyncio
import pymapreduce

@pymapreduce.remote
class CounterActor:
    def __init__(self, initial_value: int = 0):
        self.count = initial_value

    def increment(self, step: int = 1) -> int:
        self.count += step
        return self.count

    def get_value(self) -> int:
        return self.count

async def main():
    driver = await pymapreduce.DriverWrapper.init("127.0.0.1:7777")

    # Instantiate Actor on a dedicated worker
    counter = await CounterActor.remote(initial_value=100)

    # Subsequent method invocations route deterministically to the same worker instance
    await counter.increment.remote(5)
    await counter.increment.remote(10)
    final_count = await counter.get_value.remote()

    print("Actor Value:", final_count) # 115

if __name__ == "__main__":
    asyncio.run(main())
```

---

#### C. Passing Large Objects (`ObjectRef` & Zero-Copy Pipeline)
When a task produces large outputs (e.g. DataFrames, Numpy arrays, ML models), the framework stores the binary payload in the tiered **ObjectStore** and returns a lightweight `ObjectRef`.

```python
import asyncio
import pymapreduce
import pandas as pd

@pymapreduce.remote
def load_dataset() -> pd.DataFrame:
    # Heavy data loading
    return pd.DataFrame({"feature": range(1000000), "target": [1] * 1000000})

@pymapreduce.remote
def train_model(df: pd.DataFrame):
    return f"Model trained on {len(df)} rows"

async def main():
    driver = await pymapreduce.DriverWrapper.init("127.0.0.1:7777")

    # 1. load_dataset returns an ObjectRef handle
    data_ref = await load_dataset.remote()
    print("Object Reference:", data_ref.object_id)

    # 2. Pass data_ref directly into downstream task (P2P zero-copy transfer)
    result = await train_model.remote(data_ref)
    print("Training Output:", result)

    # 3. Or explicitly fetch data to driver if needed
    raw_df = await driver.get(data_ref.object_id)
    print("Fetched DataFrame Shape:", raw_df.shape)

if __name__ == "__main__":
    asyncio.run(main())
```

---

#### D. Dynamic Module Sync (`runtime_env`)
Ship custom Python scripts and modules to remote workers on the fly:
```python
driver = await pymapreduce.DriverWrapper.init(
    address="192.168.1.100:7777",
    runtime_env={"working_dir": "./my_local_modules"}
)
```

---

## 🎛️ Low-Level Job API

For granular control over task dependencies, retry policies, and timeouts:

```python
import pymapreduce

# Create custom job
job = pymapreduce.JobConfig("CustomPipeline")
job.with_max_time(120) # 120 seconds timeout limit

# Define task inputs
task = pymapreduce.TaskInput(pymapreduce.TaskKind.Map, (my_func, (arg1, arg2)))
job.add_task(task)

# Submit to cluster
results = await driver.submit(job)
```

---

## 🔬 Architecture Under The Hood

```
┌─────────────────────────────────────────────────────────────┐
│                      Python Driver (GIL)                    │
│   @remote calls -> cloudpickle serialization -> ObjectRef   │
└──────────────────────────────┬──────────────────────────────┘
                               │ PyO3 + Tokio Async Channels
┌──────────────────────────────▼──────────────────────────────┐
│                    Rust Worker Engine Core                  │
│    - Length-delimited TCP framing (up to 70MB per chunk)     │
│    - Semaphore back-pressure (physical CPU bounds)          │
│    - Tiered Object Store (Hot RAM -> Warm Disk LRU Spill)   │
└──────────────────────────────┬──────────────────────────────┘
                               │ IPC Binary Pipe (stdin/stdout)
┌──────────────────────────────▼──────────────────────────────┐
│                  Python Daemon Subprocess                   │
│    - Worker loop executing pure Python code                 │
│    - Dedicated memory space per worker process              │
│    - Exception tracebacks captured & reported to GCS        │
└─────────────────────────────────────────────────────────────┘
```

---

## 📜 License

Distributed under the MIT / Apache-2.0 License.

