pipedata 0.3

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