GitLocker: The Coding Marketplace

Description:

pipedata 0.3

pipedata
Chained operations in Python, applied to data processing.
Installation
To install with all optional dependencies:
pip install pipedata[ops]

If you only want the core functionality (building pipelines), and not
the data processing applications, then:
pip install pipedata

Examples
Chaining Data Operations
pipedata.ops provides some operations for streaming data through memory.
import json
import zipfile

import pyarrow.parquet as pq

from pipedata.core import Stream
from pipedata.ops import json_records, parquet_writer, zipped_files

data1 = [
{"col1": 1, "col2": "Hello"},
{"col1": 2, "col2": "world"},
]
data2 = [
{"col1": 3, "col2": "!"},
]

with zipfile.ZipFile("test_input.json.zip", "w") as zipped:
zipped.writestr("file1.json", json.dumps(data1))
zipped.writestr("file2.json", json.dumps(data2))

result = (
Stream(["test_input.json.zip"])
.then(zipped_files)
.then(json_records())
.then(parquet_writer("test_output.parquet"))
.to_list()
)

table = pq.read_table("test_output.parquet")
print(table.to_pydict())
#> {'col1': [1, 2, 3], 'col2': ['Hello', 'world', '!']}

Alternatively, you can construct the pipeline as a chain:
import pyarrow.parquet as pq

from pipedata.core import Chain, Stream
from pipedata.ops import json_records, parquet_writer, zipped_files

# Running this after input file created in above example
chain = (
Chain()
.then(zipped_files)
.then(json_records())
.then(parquet_writer("test_output_2.parquet"))
)
result = Stream(["test_input.json.zip"]).then(chain).to_list()
table = pq.read_table("test_output_2.parquet")
print(table.to_pydict())
#> {'col1': [1, 2, 3], 'col2': ['Hello', 'world', '!']}

Core Framework
The core framework provides the building blocks for chaining operations.
Running a stream:
from pipedata.core import Stream, ops

result = (
Stream(range(10))
.then(ops.filtering(lambda x: x % 2 == 0))
.then(ops.mapping(lambda x: x ^ 2))
.then(ops.batched(lambda x: x, 2))
.to_list()
)
print(result)
#> [(2, 0), (6, 4), (10,)]

Creating a chain and then using it, this time using the
pipe notation:
import json
from pipedata.core import Chain, Stream, ops

chain = (
Chain()
| ops.filtering(lambda x: x % 2 == 0)
| ops.mapping(lambda x: x ^ 2)
| ops.batched(lambda x: sum(x), 2)
)
print(Stream(range(10)).then(chain).to_list())
#> [2, 10, 10]
print(json.dumps(chain.get_counts(), indent=4))
#> [
#> {
#> "name": "_identity",
#> "inputs": 10,
#> "outputs": 10
#> },
#> {
#> "name": "<lambda>",
#> "inputs": 10,
#> "outputs": 5
#> },
#> {
#> "name": "<lambda>",
#> "inputs": 5,
#> "outputs": 5
#> },
#> {
#> "name": "<lambda>",
#> "inputs": 5,
#> "outputs": 3
#> }
#> ]

Similar Functionality

Python has built in functionality for building iterators

LangChain implements chained operations using its
Runnable protocol