smallpond Documentation

Smallpond is a lightweight distributed data processing framework designed for high performance and scalability. It utilizes duckdb as its core compute engine and uses parquet format for data storage on distributed file systems (such as 3FS).

Key benefits include:

• Performance: Native-level performance via DuckDB.
• Scalability: Handles PB-scale data by leveraging distributed file systems for intermediate storage, avoiding memory bottlenecks.
• Simplicity: Minimal dependencies and no requirement for long-running services, simplifying deployment and maintenance.

The DataFrame is the primary class in smallpond. It represents a partitioned dataset that is computed lazily. You define a sequence of transformations (like mapping, filtering, or repartitioning) and then trigger execution using consuming methods (like write_parquet, to_pandas, or count).

import smallpond

sp = smallpond.init()

df = sp.read_parquet("path/to/dataset/*.parquet")
df = df.repartition(10)
df = df.map("x + 1")
df.write_parquet("path/to/output")

The High-level API is centered around the DataFrame object. It is the recommended way to use Smallpond for dynamic construction of data flow graphs, execution, and result retrieval. It currently uses Ray as the backend.

A typical workflow involves initializing a session, reading data (e.g., Parquet), performing transformations like repartition or map, and writing the results.

import smallpond

sp = smallpond.init()

df = sp.read_parquet("path/to/dataset/*.parquet")
df = df.repartition(10)
df = df.map("x + 1")
df.write_parquet("path/to/output")

Smallpond allows you to specify built-in platforms or implement your own custom Platform class. By default, Smallpond attempts to auto-detect the environment.

To use a built-in platform like mpi, use the --platform flag. To use a custom platform, provide the import path to your class.

To use smallpond for high-performance data processing, follow these steps:

1. Initialize a session using smallpond.init().
2. Load data using methods like read_parquet().
3. Perform operations such as repartition() or execute SQL queries via partial_sql().
4. Save or display results using write_parquet() or to_pandas().

import smallpond

# Initialize session
sp = smallpond.init()

# Load data
df = sp.read_parquet("prices.parquet")

# Process data
df = df.repartition(3, hash_by="ticker")
df = sp.partial_sql("SELECT ticker, min(price), max(price) FROM {0} GROUP BY ticker", df)

# Save results
df.write_parquet("output/")
# Show results
print(df.to_pandas())

The Driver class is the recommended entry point for executing a LogicalPlan. It acts as a wrapper around the JobManager, automatically reading configuration from command-line arguments and passing them to the manager. You can extend the Driver by adding custom arguments using add_argument before building and running your plan.

from smallpond.execution.driver import Driver

if __name__ == "__main__":
   driver = Driver()
   # add your own arguments
   driver.add_argument("-i", "--input_paths", nargs="+")
   driver.add_argument("-n", "--npartitions", type=int, default=10)
   # build and run logical plan
   plan = my_pipeline(**driver.get_arguments())
   driver.run(plan)

To start working with smallpond, you must first initialize a session using smallpond.init().

import smallpond

sp = smallpond.init()

In smallpond, data processing pipelines are constructed by chaining Node objects together into a directed acyclic graph (DAG). Each node represents a specific operation or transformation. A typical workflow involves initializing a Context, defining a ParquetDataSet, and then wrapping these in various node types (like DataSourceNode, DataSetPartitionNode, or SqlEngineNode) before finalizing the workflow with a LogicalPlan.

# Create a global context
ctx = Context()

# Create a dataset
dataset = ParquetDataSet("path/to/dataset/*.parquet")

# Create a data source node
node = DataSourceNode(ctx, dataset)

# Partition the data
node = DataSetPartitionNode(ctx, (node,), npartitions=2)

# Create a SQL engine node to transform the data
node = SqlEngineNode(ctx, (node,), "SELECT * FROM {0}")

# Create a logical plan from the root node
plan = LogicalPlan(ctx, node)

To create a Dataset representing a collection of Parquet files, use the ParquetDataSet class by providing a glob pattern to the file paths.

   dataset = ParquetDataSet("path/to/dataset/*.parquet")

To run tasks in smallpond, you must follow a three-step workflow:

1. Initialize a RuntimeContext: This manages the environment and job identity.
2. Create a Logical Plan: Define the high-level sequence of operations.
3. Create an ExecutionPlan: Use a Planner to transform the logical plan into an executable plan based on the runtime context.

Once the ExecutionPlan is created, it can be passed to a scheduler for execution.

# create a runtime context
runtime_ctx = RuntimeContext(JobId.new(), data_root)
runtime_ctx.initialize(socket.gethostname(), cleanup_root=True)

# create a logical plan
plan = create_logical_plan()

# create an execution plan
planner = Planner(runtime_ctx)
exec_plan = planner.create_exec_plan(plan)

Install the smallpond package using pip. Supported Python versions are 3.8 to 3.12.

pip install smallpond