repository·master·Indexed 25 days ago
https://github.com/entilzha/pyfunctionalA Python library for creating data pipelines using chained functional operators. It provides an expressive API inspired by Scala, Spark, and LINQ, supporting lazy execution, parallel processing via pseq, and integration with various data sources including JSON, CSV, and SQLite3. The library categorizes operations into Streams (entry points), Transformations (lazy modifiers like map and filter), and Actions (terminal evaluations like reduce and sum).
PyFunctional operations are categorized into three types that form a data pipeline:
seq(data) or seq.json('file.json')).map, flat_map, filter, where, select).to_list, reduce, to_dict, sum, max_by).Example pipeline: seq(1, 2, 3).map(lambda x: x * 2).reduce(lambda x, y: x + y)
seq is the streammap is the transformationreduce is the action.seq(1, 2, 3).map(lambda x: x * 2).reduce(lambda x, y: x + y)PyFunctional uses lazy execution by tracking the list of transformations applied to a sequence (called its lineage). Computations are only evaluated when an action is called.
This lineage tracking allows for caching results to prevent expensive re-computation. For example, calling size() or repr() will cache the underlying sequence so that subsequent operations don'|n't re-run the entire pipeline.
To inspect the current lineage, you can access the _lineage attribute.
def times_2(x):
return 2 * x
elements = (
seq(1, 1, 2, 3, 4)
.map(times_2)
.peek(print)
.distinct()
)
elements._lineage
# Lineage: sequence -> map(times_2) -> peek(print) -> distinct
l_elements = elements.to_list()
# Prints: 1
# Prints: 1
# Prints: 2
# Prints: 3
# Prints: 4
elements._lineage
# Lineage: sequence -> map(times_2) -> peek(print) -> distinct -> cache
l_elements = elements.to_list()
# The cached result is returned so times_2 is not called and nothing is printedInstall the pyfunctional package using pip from your command line.
$ pip install pyfunctionalseq function from functional. The seq function returns an instance of functional.streams.Stream, which serves as the entry point for all data processing. You can use seq to wrap existing Python collections or use specialized stream methods like .csv() to read data from files.To use the library, import the seq function from the functional module.
from functional import seqTo parallelize "embarrassingly parallel" operations, import pseq instead of seq. This uses Python's multiprocessing to run operations in parallel.
Supported parallel operations:
map / selectfilter / filter_not / whereflat_mapPyFunctional automatically squashes chains of these operations to reduce multiprocessing overhead.
When using to_ functions to write files, you can specify compression. PyFunctional supports gzip (or gz), lzma (or xz), and bz2.
Set the compression parameter to one of these values. If not set, it defaults to None (no compression).
PyFunctional provides multiple ways to express the same logic, allowing you to choose the syntax that fits your background:
filter and map.where and select.fn style: Uses the fn library's underscore _ syntax for more concise lambdas.Example filtering transactions:
from functional import seq
from collections import namedtuple
from fn import _
Transaction = namedtuple('Transaction', 'reason amount')
transactions = [
Transaction('github', 7),
Transaction('food', 10),
Transaction('coffee', 5),
# ... other transactions
]
# Scala/Spark inspired
food_cost = seq(transactions)\n .filter(lambda x: x.reason == 'food')\n .map(lambda x: x.amount).sum()
# LINQ inspired
food_cost = seq(transactions)\n .where(lambda x: x.reason == 'food')\n .select(lambda x: x.amount).sum()
# Using fn underscore syntax
food_cost = seq(transactions).filter(_.reason == 'food').map(_.amount).sum()PyFunctional excels at complex data manipulations like grouping, reducing by key, and joining different data sources.
reduce_by_key: Useful for word counts or frequency analysis.group_by: Groups elements by a key.inner_join: Joins two streams based on common keys.# Word count example
words = 'I dont want to believe I want to know'.split(' ')
seq(words).map(lambda word: (word, 1)).reduce_by_key(lambda x, y: x + y)
# Joining JSON users with JSONL messages
users = seq.json('examples/users.json')
messages = seq.jsonl('examples/chat_logs.jsonl')
message_tuples = messages.group_by(lambda m: m['user'])
data = users.inner_join(message_tuples)By default, functions like first() or last() wrap their result in a Sequence if the returned element is itself an iterable. To get the raw element instead, use the no_wrap=True option.
This option can be passed to:
first(no_wrap=True)last(no_wrap=True)head(no_wrap=True)head_option(no_wrap=True)last_option(no_wrap=True)seq(..., no_wrap=True)Sequence() constructor>>> s = seq(list(), list())
>>> type(s.first())
<class 'functional.pipeline.Sequence'>
>>> type(s.first(no_wrap=True))
<class 'list'>
>>> type(seq([list(), list()], no_wrap=True).last())
<class 'list'>