5 ms·

`riko` is pure python stream processing library for analyzing and processing streams of structured data. It's modeled after Yahoo! Pipes [1] and was originally

by reubano 10y ago

`riko` is pure python stream processing library for analyzing and processing streams of structured data. It's modeled after Yahoo! Pipes [1] and was originally a fork of pipe2py [2]. It has both synchronous and asynchronous (via twisted) APIs, and supports parallel execution (via multiprocessing). Out of the box, `riko` can read csv/xml/json/html files; create text and data based flows via modular pipes; parse and extract RSS/ATOM feeds; and bunch of other neat things. You can think of `riko` as a poor man's Spark/Storm... stream processing made easy! Feedback welcome so let me know what you think! Resources: FAQ [3], cookbook [4], and ipython notebook [5] Quickie Demo: >>> from riko.modules import fetch >>> >>> stream = fetch.pipe(conf={'url': 'https://news.ycombinator.com/rss'}) >>> item = next(stream) >>> item['title'], item['link'] ('Master Plan, Part Deux', 'https://www.tesla.com/blog/master-plan-part-deux') [1] https://web.archive.org/web/20150930021241/http://pipes.yahoo.com/pipes/ https://web.archive.org/web/20150930021241/http://pipes.yaho... [2] https://github.com/ggaughan/pipe2py/ https://github.com/ggaughan/pipe2py/ [3] https://github.com/nerevu/riko/blob/master/docs/FAQ.rst https://github.com/nerevu/riko/blob/master/docs/FAQ.rst [4] https://github.com/nerevu/riko/blob/master/docs/COOKBOOK.rst https://github.com/nerevu/riko/blob/master/docs/COOKBOOK.rst [5] http://nbviewer.jupyter.org/github/nerevu/riko/blob/master/examples/usage.ipynb http://nbviewer.jupyter.org/github/nerevu/riko/blob/master/e...

olviko 10y ago
Nice project. I wrote something similar in C# long time ago [1]. Mostly to monitor job feeds and craigslist [2] :-) It supports RSS and Atom, async, various filters, deduplication, etc Yahoo Pipes was a nice project, but as its popularity grew, it started getting blocked more and more. It was also hard to build and maintain pipelines with more than a few steps. [1] https://github.com/olviko/RssPercolator https://github.com/olviko/RssPercolator [2] https://github.com/olviko/RssPercolator/blob/master/RssPercolator.Client/Program.cs https://github.com/olviko/RssPercolator/blob/master/RssPerco...
- reubano 10y ago
  Cool! I actually starred your project last year. Never really got around to looking under the hood though. How do you handle the "multiple destinations" part? In python you can do it with a coroutines [1, 2] implementation (push based). I avoided that, since it coroutines have their own form of callback hell, and decided to implement a generator api (pull based) [3, 4]. But since generators can only be "pulled" into one destination, you have to copy a stream (subsequently converting it into a list) if you want more than one destination [5]. This works fine if the data can fit in memory, but if it can't then you're out of luck! [1] http://www.dabeaz.com/coroutines/copipe.py http://www.dabeaz.com/coroutines/copipe.py [2] http://www.dabeaz.com/coroutines/ http://www.dabeaz.com/coroutines/ [3] http://www.dabeaz.com/generators/retuple.py http://www.dabeaz.com/generators/retuple.py [4] http://www.dabeaz.com/generators http://www.dabeaz.com/generators [5] https://github.com/nerevu/riko/blob/master/riko/modules/split.py#L15-L18 https://github.com/nerevu/riko/blob/master/riko/modules/spli...
  - olviko 10y ago
    Can’t really remember, sorry, that code was built for processing RSS feeds and data size was never an issue. I will take a look when get some free time… But, I see what you mean. I had to deal with similar issues in commercial projects and the "pull" model (generators in Python ~ "yield return" in C#) almost never a good idea, especially when you have to have concurrent consumers. While callbacks are hard to combine, in C# it can be nicely abstracted with “async/await”, not sure how it is handled in Python, I stopped using it around 2.5
    - greenshackle 10y ago
      Python 3.5 introduced the async/await syntax, I don't do C# but at a glance it's the same. I've been working on a similar project and I've also found the push model easier.
      - reubano 10y ago
        Care to share your code? I'd like to see a proper push implementation.
- rubber_duck 10y ago
  Maybe I'm not seeing something but wouldn't this be simpler with http://reactivex.io/ http://reactivex.io/ (Rx.NET specifically) ? Just implement IObservable that pushes RSS events and then use Rx Filter/Select/Merge/GroupBy to filter/join/synchronize/whatever
  - olviko 10y ago
    Maybe. Don't think any of this existed when I need it... 1. The code is tiny with 90% of it dealing with RSS parsing and filtering. Using RX.NET wouldn't really simplify anything. 2. I wanted a library that I can integrate into my apps and run locally to avoid throttling, robots.txt and other BS Yahoo Pipes was suffering from. I am also not a huge fan of RX... to put it mildly
    - rubber_duck 10y ago
      I think I've used Rx way back when in 2009 (there was no TPL in silverlight so you had to use callbacks and events for continuations which was insane, so I used Rx) so it's been around. Mind sharing why you don't like it ? I personally like how it allows me to express complex high level operations cleanly. For example - I have a observable configuration variable that can come from different sources and the source change dynamically. I need to listen to latest source until a new one becomes active - in Rx I only need to push the new source trough IObservable<IObservable<ConfigurationValue>> and then use http://reactivex.io/documentation/operators/switch.html http://reactivex.io/documentation/operators/switch.html which returns IObservable<ConfigurationValue> which will push values from the latest source - Rx will handle unsubscribing from previous active source, synchronizing state and making sure everything is thread safe. And there are a bunch of operators like this that would be tedious and hard implement correctly with all the edge cases in a thread safe way - and here they are abstracted in to high level operators.
alexchamberlain 10y ago
This is an implementation of "Flow Based Programming", right? Its a programming paradigm invented before its time IMHO; perfect for a world of streaming data.
- reubano 10y ago
  > This is an implementation of "Flow Based Programming", right? pretty much.... just without a GUI. My inspiration was method chaining [1] and the first implementation was this [2]. [1] http://martinfowler.com/articles/collection-pipeline/ http://martinfowler.com/articles/collection-pipeline/ [2] http://stackoverflow.com/questions/12172934/method-chaining-in-python/28232547#28232547 http://stackoverflow.com/questions/12172934/method-chaining-...
  - asavinov 10y ago
    I am currently working on an alternative approach which is based on a column-oriented paradigm for in-stream analytics as opposed to flow-orientation. It is an adaptation of DataCommandr (http://conceptoriented.com http://conceptoriented.com) which is a column-oriented approach to data processing. Unfortunately, StreamCommandr is not yet available but it would be interesting to compare advanatages and drawbacks of these two paradigms.
    - reubano 10y ago
      I didn't see any code examples so it's a bit difficult to figure out what's going on. Are there any open source column-oriented programming implementations?
      - asavinov 10y ago
        Actually, it is a novel approach so I am not aware of anything similar. Initially, I implemented it for data wrangling (various potentially complex transformations: https://en.wikipedia.org/wiki/Data_wrangling https://en.wikipedia.org/wiki/Data_wrangling) and it can be found here: https://bitbucket.org/conceptoriented/dce-java https://bitbucket.org/conceptoriented/dce-java. The main point is that instead of defining a flow (graph) of tables we define a flow (graph) of columns. It can be used where Apache Spark is used. StreamCommandr essentially relies on the same principles and changes only external API and some internal processing. The idea is that any data table is like a stream of records so that we add new records and delete outdated records. Simultaniously, we evaluate other columns and tables by performaing potentially complex computations which are difficult to do in a record flow.

by reubano 10y ago

olviko 10y ago

Nice project. I wrote something similar in C# long time ago [1]. Mostly to monitor job feeds and craigslist [2] :-) It supports RSS and Atom, async, various filters, deduplication, etc Yahoo Pipes was a nice project, but as its popularity grew, it started getting blocked more and more. It was also hard to build and maintain pipelines with more than a few steps. [1] https://github.com/olviko/RssPercolator https://github.com/olviko/RssPercolator [2] https://github.com/olviko/RssPercolator/blob/master/RssPercolator.Client/Program.cs https://github.com/olviko/RssPercolator/blob/master/RssPerco...

reubano 10y ago
Cool! I actually starred your project last year. Never really got around to looking under the hood though. How do you handle the "multiple destinations" part? In python you can do it with a coroutines [1, 2] implementation (push based). I avoided that, since it coroutines have their own form of callback hell, and decided to implement a generator api (pull based) [3, 4]. But since generators can only be "pulled" into one destination, you have to copy a stream (subsequently converting it into a list) if you want more than one destination [5]. This works fine if the data can fit in memory, but if it can't then you're out of luck! [1] http://www.dabeaz.com/coroutines/copipe.py http://www.dabeaz.com/coroutines/copipe.py [2] http://www.dabeaz.com/coroutines/ http://www.dabeaz.com/coroutines/ [3] http://www.dabeaz.com/generators/retuple.py http://www.dabeaz.com/generators/retuple.py [4] http://www.dabeaz.com/generators http://www.dabeaz.com/generators [5] https://github.com/nerevu/riko/blob/master/riko/modules/split.py#L15-L18 https://github.com/nerevu/riko/blob/master/riko/modules/spli...
- olviko 10y ago
  Can’t really remember, sorry, that code was built for processing RSS feeds and data size was never an issue. I will take a look when get some free time… But, I see what you mean. I had to deal with similar issues in commercial projects and the "pull" model (generators in Python ~ "yield return" in C#) almost never a good idea, especially when you have to have concurrent consumers. While callbacks are hard to combine, in C# it can be nicely abstracted with “async/await”, not sure how it is handled in Python, I stopped using it around 2.5
  - greenshackle 10y ago
    Python 3.5 introduced the async/await syntax, I don't do C# but at a glance it's the same. I've been working on a similar project and I've also found the push model easier.
    - reubano 10y ago
      Care to share your code? I'd like to see a proper push implementation.
rubber_duck 10y ago
Maybe I'm not seeing something but wouldn't this be simpler with http://reactivex.io/ http://reactivex.io/ (Rx.NET specifically) ? Just implement IObservable that pushes RSS events and then use Rx Filter/Select/Merge/GroupBy to filter/join/synchronize/whatever
- olviko 10y ago
  Maybe. Don't think any of this existed when I need it... 1. The code is tiny with 90% of it dealing with RSS parsing and filtering. Using RX.NET wouldn't really simplify anything. 2. I wanted a library that I can integrate into my apps and run locally to avoid throttling, robots.txt and other BS Yahoo Pipes was suffering from. I am also not a huge fan of RX... to put it mildly
  - rubber_duck 10y ago
    I think I've used Rx way back when in 2009 (there was no TPL in silverlight so you had to use callbacks and events for continuations which was insane, so I used Rx) so it's been around. Mind sharing why you don't like it ? I personally like how it allows me to express complex high level operations cleanly. For example - I have a observable configuration variable that can come from different sources and the source change dynamically. I need to listen to latest source until a new one becomes active - in Rx I only need to push the new source trough IObservable<IObservable<ConfigurationValue>> and then use http://reactivex.io/documentation/operators/switch.html http://reactivex.io/documentation/operators/switch.html which returns IObservable<ConfigurationValue> which will push values from the latest source - Rx will handle unsubscribing from previous active source, synchronizing state and making sure everything is thread safe. And there are a bunch of operators like this that would be tedious and hard implement correctly with all the edge cases in a thread safe way - and here they are abstracted in to high level operators.

alexchamberlain 10y ago

This is an implementation of "Flow Based Programming", right? Its a programming paradigm invented before its time IMHO; perfect for a world of streaming data.

reubano 10y ago
> This is an implementation of "Flow Based Programming", right? pretty much.... just without a GUI. My inspiration was method chaining [1] and the first implementation was this [2]. [1] http://martinfowler.com/articles/collection-pipeline/ http://martinfowler.com/articles/collection-pipeline/ [2] http://stackoverflow.com/questions/12172934/method-chaining-in-python/28232547#28232547 http://stackoverflow.com/questions/12172934/method-chaining-...
- asavinov 10y ago
  I am currently working on an alternative approach which is based on a column-oriented paradigm for in-stream analytics as opposed to flow-orientation. It is an adaptation of DataCommandr (http://conceptoriented.com http://conceptoriented.com) which is a column-oriented approach to data processing. Unfortunately, StreamCommandr is not yet available but it would be interesting to compare advanatages and drawbacks of these two paradigms.
  - reubano 10y ago
    I didn't see any code examples so it's a bit difficult to figure out what's going on. Are there any open source column-oriented programming implementations?
    - asavinov 10y ago
      Actually, it is a novel approach so I am not aware of anything similar. Initially, I implemented it for data wrangling (various potentially complex transformations: https://en.wikipedia.org/wiki/Data_wrangling https://en.wikipedia.org/wiki/Data_wrangling) and it can be found here: https://bitbucket.org/conceptoriented/dce-java https://bitbucket.org/conceptoriented/dce-java. The main point is that instead of defining a flow (graph) of tables we define a flow (graph) of columns. It can be used where Apache Spark is used. StreamCommandr essentially relies on the same principles and changes only external API and some internal processing. The idea is that any data table is like a stream of records so that we add new records and delete outdated records. Simultaniously, we evaluate other columns and tables by performaing potentially complex computations which are difficult to do in a record flow.