<commit_before>Streams
=======

Do not use.  This is an exploratory project to learn more about what it takes
to build typical real time systems.

Other projects you should probably consider instead:

1.  [RxPY](https://github.com/ReactiveX/RxPY)
2.  [Flink](https://flink.apache.org/)


Basic Explanation
-----------------

This implements a trivial subset of an API commonly found in streaming data
systems.  You can have a Stream and can create various other streams from that
stream using common modifiers like map, filter, scan, etc..  Eventually you
create a sink that consumes results.

```python
from streams import Stream

source = Stream()

L = []
stream = source.map(inc).scan(add)
stream.sink(L.append)
stream.sink(print)
```

Now as you feed data into the source all of the operations trigger as necessary


```python
>>> for i in range(3):
...     source.emit(i)
1
3
7

>>> L
[1, 3, 7]
```

You can use the typical map/filter/scan syntax.  Everything can have
multiple subscribers at any point in the stream.

Additionally everything responds to backpressure, so if the sink blocks the
source will block (although you can add in buffers if desired).  Additionally
everything supports asynchronous workloads with Tornado coroutines, so you can
do async/await stuff if you prefer (or gen.coroutine/yield in Python 2).

```python
async def f(result):
    ... do non-blocking stuff with result ...

stream.sink(f)

for i in range(10):
    await source.emit(i)
```

This means that if the sinks can't keep up then the sources will stop pushing
data into the system.  This is useful to control buildup.


Dask
----

Everything above runs with normal Python in the main thread or optionally in a
Tornado event loop.  Alternatively this library plays well with Dask.  You can
scatter data to the cluster, map and scan things up there, gather back, etc..

```python
source.to_dask().scatter().map(func).scan(func).gather().sink(...)
```

Less Trivial Example
--------------------

```python
source = Source()
output = open('out')

s = source.map(json.loads)        # Parse JSON data
          .timed_window(0.050)    # Batch into 50ms batches
          .filter(len)            # Remove any batches that didn't have data
          .to_dask().scatter()    # Send to cluster
          .map(pd.DataFrame)      # Convert to pandas dataframes on the cluster
          .map(pd.DataFrame.sum)  # Sum rows of each batch
          .scan(add)              # Maintain running sum of all data
          .gather()               # Collect results back to local machine
          .map(str)               # Convert to string
          .sink(output.write)     # Write to file

from kafka_library import KafkaReader

topic = KafkaReader('data')

while True:
    for line in topic:
        source.emit(line)
```<commit_msg>add whats missing section to readme<commit_after>Streams
=======

Do not use.  This is an exploratory project to learn more about what it takes
to build typical real time systems.

Other projects you should probably consider instead:

1.  [RxPY](https://github.com/ReactiveX/RxPY)
2.  [Flink](https://flink.apache.org/)


Basic Explanation
-----------------

This implements a trivial subset of an API commonly found in streaming data
systems.  You can have a Stream and can create various other streams from that
stream using common modifiers like map, filter, scan, etc..  Eventually you
create a sink that consumes results.

```python
from streams import Stream

source = Stream()

L = []
stream = source.map(inc).scan(add)
stream.sink(L.append)
stream.sink(print)
```

Now as you feed data into the source all of the operations trigger as necessary


```python
>>> for i in range(3):
...     source.emit(i)
1
3
7

>>> L
[1, 3, 7]
```

You can use the typical map/filter/scan syntax.  Everything can have
multiple subscribers at any point in the stream.

Additionally everything responds to backpressure, so if the sink blocks the
source will block (although you can add in buffers if desired).  Additionally
everything supports asynchronous workloads with Tornado coroutines, so you can
do async/await stuff if you prefer (or gen.coroutine/yield in Python 2).

```python
async def f(result):
    ... do non-blocking stuff with result ...

stream.sink(f)

for i in range(10):
    await source.emit(i)
```

This means that if the sinks can't keep up then the sources will stop pushing
data into the system.  This is useful to control buildup.


Dask
----

Everything above runs with normal Python in the main thread or optionally in a
Tornado event loop.  Alternatively this library plays well with Dask.  You can
scatter data to the cluster, map and scan things up there, gather back, etc..

```python
source.to_dask().scatter().map(func).scan(func).gather().sink(...)
```

Less Trivial Example
--------------------

```python
source = Source()
output = open('out')

s = source.map(json.loads)        # Parse JSON data
          .timed_window(0.050)    # Batch into 50ms batches
          .filter(len)            # Remove any batches that didn't have data
          .to_dask().scatter()    # Send to cluster
          .map(pd.DataFrame)      # Convert to pandas dataframes on the cluster
          .map(pd.DataFrame.sum)  # Sum rows of each batch
          .scan(add)              # Maintain running sum of all data
          .gather()               # Collect results back to local machine
          .map(str)               # Convert to string
          .sink(output.write)     # Write to file

from kafka_library import KafkaReader

topic = KafkaReader('data')

while True:
    for line in topic:
        source.emit(line)
```


What's missing?
---------------

This is still a toy library.  It has never been used for anything.  So
presumably many things are wrong.  I've tried to build a simple system that can
grow if use cases arrive.  Here are some obvious things that are missing:

1.  A lot of API.  I recommend looking at the Rx or Flink APIs to get a sense
    of what people often need.
2.  Integration to collections like lists, numpy arrays, or pandas dataframes.
    For example we should be able to think about streams of lists of things.
    In this case `seq_stream.map(func)` would apply the function across every
    element in the constituent lists.
3.  Thinking about time.  It would be nice to be able to annotate elements with
    things like event and processing time, and have this information pass
    through operations like map
4.  Multi-stream operations like zip and joins
5.  More APIs for common endpoints like Kafka
{'fraction_non_alphanumeric': 0.093764293337399, 'fraction_numerical': 0.006250952889159933, 'mean_word_length': 2.9210998206814107, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3120738', 'n_tokens_mistral': 1884, 'n_tokens_neox': 1792, 'n_words': 844}