当前位置: 首页 > 软件库 > 程序开发 > 工作流引擎 >

Riko

Python 流处理引擎
授权协议 MIT
开发语言 Python
所属分类 程序开发、 工作流引擎
软件类型 开源软件
地区 不详
投 递 者 慕佑运
操作系统 跨平台
开源组织
适用人群 未知
 软件概览

Riko是一款Python 流处理引擎,类似Yahoo Pipes。采用纯python开发,用于分析处理结构化数据流。拥有同步和异步APIs,同时也支持并行RSS feeds。Riko也支持字符终端界面。

功能特性:

  • 可读取csv/xml/json/html文件。

  • 通过模块化的管道可创建文本流和数据流。

  • 可解析、处理、提取RSS/Atom feeds。

  • 可创建强大的混合型APIs和maps。

  • 支持并行处理。

使用示例代码:

>>> ### Create a SyncPipe flow ###
>>> #
>>> # `SyncPipe` is a convenience class that creates chainable flows
>>> # and allows for parallel processing.
>>> from riko.collections.sync import SyncPipe
>>>
>>> ### Set the pipe configurations ###
>>> #
>>> # Notes:
>>> #   1. the `detag` option will strip all html tags from the result
>>> #   2. fetch the text contained inside the 'body' tag of the hackernews
>>> #      homepage
>>> #   3. replace newlines with spaces and assign the result to 'content'
>>> #   4. tokenize the resulting text using whitespace as the delimeter
>>> #   5. count the number of times each token appears
>>> #   6. obtain the raw stream
>>> #   7. extract the first word and its count
>>> #   8. extract the second word and its count
>>> #   9. extract the third word and its count
>>> url = 'https://news.ycombinator.com/'
>>> fetch_conf = {
...     'url': url, 'start': '<body>', 'end': '</body>', 'detag': True}  # 1
>>>
>>> replace_conf = {
...     'rule': [
...         {'find': '\r\n', 'replace': ' '},
...         {'find': '\n', 'replace': ' '}]}
>>>
>>> flow = (
...     SyncPipe('fetchpage', conf=fetch_conf)                           # 2
...         .strreplace(conf=replace_conf, assign='content')             # 3
...         .stringtokenizer(conf={'delimiter': ' '}, emit=True)         # 4
...         .count(conf={'count_key': 'content'}))                       # 5
>>>
>>> stream = flow.output                                                 # 6
>>> next(stream)                                                         # 7
{"'sad": 1}
>>> next(stream)                                                         # 8
{'(': 28}
>>> next(stream)                                                         # 9
{'(1999)': 1}
 相关资料
  • 相反,如果我从web服务器收集数据,为什么不直接使用相同的节点进行事件处理呢?这些操作已经由负载均衡器分布在节点上,我在web服务器上使用负载均衡器。我可以在相同的JVM实例上创建执行器,并将事件从web服务器异步发送到执行器,而不涉及任何额外的IO请求。我还可以监视web服务器中的执行器,并确保执行器处理了事件(至少一次或恰好一次处理保证)。通过这种方式,管理我的应用程序将容易得多,而且由于不需

  • 问题内容: 在C#中,在处理流对象时,几乎总是使用该模式。例如: 通过使用该块,我们确保代码块执行后立即在流上调用dispose。 我知道Java没有关键字的等效项,但是我的问题是,当使用Java中的对象时,是否需要做任何内务处理以确保将其处置?我在看这个代码示例,但我发现他们什么也不做。 我只是想知道Java在处理处置流方面的最佳实践是什么,或者它足以让垃圾收集器处理它。 问题答案: 通常,您必

  • 让我们考虑一个<代码>父< /代码>类,它只包含一个<代码>整数< /代码>属性。我用一个空变量创建了6个父类对象。然后我将这些对象添加到列表中。 我想通过属性的值检索相应的对象。我使用了Java8流。 但是我得到了,所以我编辑了代码: 但是如果任何对象为null,我想抛出一个异常。如果列表中没有对象为null,那么我想从列表中检索相应的对象。 如何使用Java 8 Streams使用一条语句实现

  • 我在spark streaming应用程序中看到一些失败的批处理,原因是与内存相关的问题,如 无法计算拆分,找不到块输入-0-1464774108087

  • 流处理和传统消息处理的基本区别是什么?正如人们所说,kafka是流处理的好选择,但本质上,kafka是一个类似于ActivMQ、RabbitMQ等的消息传递框架。 为什么我们通常不说ActiveMQ也适合流处理呢。 消费者消费消息的速度是否决定了它是否是流?

  • null 其中lambda1、2等是条件检查函数,例如 但不知什么原因对我不起作用,也许还有其他方法?正如我从文档(https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/side_output.html)中了解到的,OutputTag用于创建标记为tag的附加消息。还是我错了?

  • 以下代码正在为空的属性抛出NPE。class Person有属性:string:name,Integer:age,Integer:salary此处可以为空。我想创建一个工资列表。 在这里,我必须在结果列表中保留空值。null不能替换为0。

  • 配置好DispatcherServlet以后,开始有请求会经过这个DispatcherServlet。此时,DispatcherServlet会依照以下的次序对请求进行处理: 首先,搜索应用的上下文对象WebApplicationContext并把它作为一个属性(attribute)绑定到该请求上,以便控制器和其他组件能够使用它。属性的键名默认为DispatcherServlet.WEB_APPL