tanbih-pipeline 0.7.6

特点

• 每封邮件至少有一次确认
• 通过消费者群体横向扩展
• 在部署中控制流程，开发一次，在任何地方使用它
• 提供文件和内存输入/输出的可测试性

参数

• kind-指定管道的下划线技术，例如KAFKA或RabbitMQ
• MEM—基于内存的队列（适用于unittests）
• 基于文件的队列（适用于开发和集成测试）

发电机

生成器将在我们的管道中开发数据源时使用。消息来源 会产生没有输入的输出。爬虫可以看作是发电机。在

>>>frompipelineimportGenerator,Message>>>>>>classMyGenerator(Generator):...defgenerate(self):...foriinrange(10):...yield{'id':i}>>>>>>generator=MyGenerator('generator','0.1.0',description='simple generator')>>>generator.parse_args("--kind MEM --out-topic test".split())>>>generator.start()>>>[r.get('id')forringenerator.destination.results][0,1,2,3,4,5,6,7,8,9]

处理器

处理器用于处理输入。将进行修改。处理器 可以为每个输入产生一个输出，或者没有输出。在

分路器

当写入多个输出时，将使用拆分器。它需要一个函数 根据处理消息生成输出主题，并在编写输出时使用它。在

>>>frompipelineimportSplitter,Message>>>>>>classMySplitter(Splitter):...defget_topic(self,msg):...return'{}-{}'.format(self.destination.topic,msg.get('id'))......defprocess(self,msg):...msg.update({...'processed':True,...})...returnNone>>>>>>splitter=MySplitter('splitter','0.1.0',description='simple splitter')>>>config={'data':[{'id':1}]}>>>splitter.parse_args("--kind MEM --in-topic test --out-topic test".split(),config=config)>>>splitter.start()>>>[r.get('id')forrinsplitter.destinations['test-1'].results][1]

使用

选择要从中生成子类的生成器、处理器或拆分器。在

贡献

使用预提交运行black和flake8

学分

张一凡（yzhang at哈佛商学院教育质量保证)在