演示基于spark的应用流_如何基于高级功能为您的应用选择流处理器？

演示基于spark的应用流

通过Miyuru Dayarathna和Srinath Perera

介绍

流处理使用户能够查询连续的数据流&＃xff0c;并在从接收时间算起的短时间内快速检测条件。 检测时间段可能从几毫秒到几分钟不等。 例如&＃xff0c;通过流处理&＃xff0c;您可以通过查询来自温度传感器的数据流并检测温度何时达到冰点来接收警报。 &＃xff08;如果您是本主题的新手&＃xff0c;请阅读流处理的温和介绍 。&＃xff09;

如果要构建一个处理流数据并做出实时决策的应用程序&＃xff0c;则有两种选择&＃xff1a;使用工具或自行构建。 答案取决于用例的复杂性&＃xff0c;可伸缩性要求以及可靠性和容错要求。

如果您想自己构建应用程序&＃xff0c;则将事件放置在消息代理主题&＃xff08;例如ActiveMQ&＃xff0c;RabbitMQ或Kafka&＃xff09;中&＃xff0c;然后编写代码以接收来自代理主题的事件&＃xff08;它们成为您的流&＃xff09;&＃xff0c;最后将结果发布回来给经纪人。 完成上述三个步骤的代码在流处理领域中称为参与者。

但是&＃xff0c;与其从头开始编写上述方案&＃xff0c;不如使用流处理框架来节省时间。 通过事件流处理器&＃xff0c;您可以使用Streaming SQL编写逻辑&＃xff0c;并处理其余部分。

您既可以将事件直接发送到流处理器&＃xff0c;也可以通过代理发送事件。 事件流处理器将通过收集数据&＃xff0c;将数据交付给每个参与者&＃xff0c;确保它们以正确的顺序运行&＃xff0c;收集结果&＃xff0c;在负载过大时进行扩展以及处理故障来完成艰苦的工作。

下一个问题是“您应该使用哪个流处理器&＃xff1f;”。 我有好消息&＃xff0c;也有坏消息。 好消息是有那么多&＃xff08;请参阅Quora问题&＃xff1a; 那里最好的流处理解决方案是什么&＃xff1f; &＃xff09;&＃xff0c;因此您有很多选择。 坏消息是&＃xff0c;有很多&＃xff0c;您可能会面临选择的悖论 。

考虑太多

不同的流处理系统支持不同的功能。

1. 使用Message Broker提取数据
2. 使用流式SQL编写查询
3. 流处理API和查询编写环境
4. 可靠性&＃xff0c;高可用性&＃xff08;HA&＃xff09;和最小HA
5. 通过拖放类型的图形用户界面&＃xff08;GUI&＃xff09;实现业务用户友好性
6. 流机器学习
7. 消息处理保证
8. 乱序事件
9. 大型系统性能&＃xff08;可伸缩性&＃xff0c;处理大窗口&＃xff09;

这是很多功能。 我们正在努力决定如何考虑19个流处理器中的所有这些方面。

想法&＃xff1a;“平均产品”

考虑了一段时间后&＃xff0c;我们提出了两个想法。

首先是必须具备vs.可选功能。 必备功能是大多数用户最终将在流处理应用程序中使用的功能&＃xff0c;因此是必需的。 其余是可选功能。

那么第二个想法是“平均产品” 。 我们将普通产品定义为假设产品&＃xff0c;该产品仅具有市场上超过2/3的流处理器支持的功能。

然后&＃xff0c;您可以评估每个产品与平均产品的对比。 如果该产品具有除普通产品之外的功能&＃xff0c;则为正&＃xff1b;而如果该产品缺少功能&＃xff0c;则与普通产品相反&＃xff0c;则为负。

给出的平均产品是将每个产品与平均值进行比较和对比的参考点。

必备功能

在调查了19个流处理器之后&＃xff0c;我们决定遵循以下四个必备功能。

1. 支持使用Message Broker进行数据提取
2. 流式SQL
3. 流处理API和查询编写环境
4. 可靠性&＃xff0c;高可用性&＃xff08;HA&＃xff09;和最小HA

我们已经在InfoQ文章中详细讨论了以上四个必备功能&＃xff0c;即如何为您的应用选择流处理器 &＃xff1f;

根据高级选择&＃xff08;可选功能&＃xff09;

本文讨论了可选功能&＃xff0c;以及如何使用通用产品的概念来评估它们。

根据我们的定义&＃xff0c;一般的Stream处理产品具有以下功能。 换句话说&＃xff0c;⅔个或更多产品具有以下功能。

• 该产品可以以分布式方式运行并具有可扩展性
• 支持至少一次消息传递保证
• 为节点故障提供容错能力
• 支持定期检查点和恢复
• 支持背压
• 支持转换&＃xff0c;聚合&＃xff0c;关联&＃xff0c;时间窗口运算符
• 支持调试
• 支持多个数据源和接收器
• 支持编写自定义运算符
• 支持监视正在运行的系统

您可以通过以下方式评估特定的流处理&＃xff1a;首先选择与应用程序相关的可选功能&＃xff0c;然后根据所选产品在平均产品中的可选功能进行比较。

让我们讨论每个可选功能以及不同用例需要什么级别的支持。

可选功能

时间观念

流处理始终是分布式应用程序。 首先&＃xff0c;从运行在一台或多台计算机上的传感器收集数据&＃xff0c;然后将其传递到流处理器&＃xff0c;进行处理&＃xff0c;然后将结果传递到一台或多台其他计算机。

因此&＃xff0c;在系统中的所有节点上都没有单一的时间概念或时钟&＃xff08;这在分布式系统中是众所周知的结果&＃xff09;。 时间有多种概念&＃xff0c;用户可以根据自己的应用选择它们。

流处理器的时间概念分为三种&＃xff1a;事件时间&＃xff0c;流时间和处理时间。

• 活动时间-事件实际产生的时间
• 流时间-流处理器将数据接收到平台中的时间
• 处理时间-时间流处理器处理数据的时间

对于处理对时间顺序敏感的操作的流应用程序&＃xff0c;时间的概念变得非常重要。

不同的流处理器支持不同级别的时间。 有些具有处理事件时间和处理时间的能力。 有些只处理流时间。 有些会处理所有三遍。 例如&＃xff0c;如果上述警报生成应用程序需要使用基于事件时间的时间窗口并在特定时间范围&＃xff08;例如&＃xff0c;下午5点至上午8点&＃xff09;内对特定字段进行比较&＃xff0c;则流处理器仅使用流时间不能用于此目的。 这是因为如果将流时间与Windows一起使用&＃xff0c;则可能会引发不正确的警报。 因此&＃xff0c;您应根据应用程序所需的时间排序支持级别选择流处理器。

区分事件时间和流时间的流处理器必须处理乱序事件的正确性问题。 这是因为&＃xff0c;事件之间在传输到流处理器的接收器时经常会发生乱序。 流中的乱序事件是由于多种原因而产生的&＃xff0c;例如操作员并行化&＃xff0c;网络延迟和异步流的合并。 通常使用乱序处理技术来处理这种情况。 乱序处理的细节将在本文后面的小节中详细讨论。

业务用户友好度

支持业务用户构建流处理应用程序可提高流处理器的可用性。 业务用户是需要针对业务用例使用流分析但对流处理解决方案的基础技术方面没有很好的技术知识的人。 业务用户友好性使开发人员和业务用户都可以设计&＃xff0c;部署和自定义他们的用例。

需要这种业务用户友好性的示例是分布在不同地理位置的业务。 在这种用例中&＃xff0c;每个地理位置的运营中心将使用不同的变量执行相同的过程。 在这种情况下&＃xff0c;以抽象的方式设计流程并为每个地理位置定制流程可以大大简化用例。

有几种示例方法&＃xff0c;其中流处理器实现了业务用户友好性。 一种方法是通过使用电子表格进行流处理。 尽管业务用户&＃xff08;例如&＃xff0c;领域专家&＃xff09;并不总是程序员&＃xff0c;但他们通常熟悉电子表格处理器&＃xff0c;因为据报道&＃xff0c;全球有超过5亿人使用电子表格 。 电子表格已用于可视化实时流&＃xff0c;实时编程以计算新的流以及导出要在服务器上运行的计算&＃xff0c;以便可以与其他用户共享计算并在电子表格的整个生命周期内持续存在。 Hirzel撰写的论文“用于无边界Windows和分区的流处理电子表格”是is方法的一个示例&＃xff0c;尽管就我们所知&＃xff0c;还没有流处理器支持这种方法。

另一个示例是在WSO2流处理器 &＃xff08;WSO2 SP&＃xff09;中使用模板。 WSO2 SP的业务规则管理器允许将查询指定为模板&＃xff08;即框架&＃xff09;。 精通编写流处理查询的开发人员将决定应将哪些值保留为变量。 在如何编写流查询方面知识有限或根本不了解的业务用户可以使用框架并创建应用程序。 例如&＃xff0c;假设我们要编写一个查询来检测车辆何时超过速度限制。 但是&＃xff0c;业务用户可以根据查询的部署位置来决定速度限制。 模板包括业务用户可以填写以更改查询行为的表单。 但是&＃xff0c;在保留相同查询结构的同时&＃xff0c;不同的参数可能会在组织和业务场景之间发生变化。 使用Business Rules Manager可以使用查询模板快速生成正在运行的查询&＃xff0c;并为参数分配值。

如前所述&＃xff0c;拖放类型的图形用户界面&＃xff08;GUI&＃xff09;允许通过使用图标和属性窗口封装查询逻辑来实现应用程序。 例如&＃xff0c;一旦用户将流组件拖放到拖放GUI中&＃xff0c;他就可以使用流组件的关联属性窗口添加属性名称及其类型。

我们调查的流处理器中只有32&＃xff05;至少为业务用户提供了一些支持&＃xff0c;以开发流处理应用程序。 提供预构建的业务应用程序和业务用户工具还可以帮助业务用户使用流处理器系统快速转换其业务流程。 在接受调查的处理器中&＃xff0c;只有3个提供了此类预构建的业务应用程序。

集成到机器学习

机器学习&＃xff08;ML&＃xff09;从数据中学习&＃xff0c;并通过编写算法来解决难以解决的问题。 例如&＃xff0c;机器学习模型可以用于检测欺诈活动。 可以将分类器与警报生成应用程序结合使用&＃xff0c;以更准确地生成警报&＃xff0c;而不是使用诸如大于比较之类的简单标准。

大多数新用例将以某种方式结合机器学习。 因此&＃xff0c;我们需要流处理器来支持机器学习模型。 但是&＃xff0c;将ML算法集成到流处理应用程序中并不是一件容易的事。

传统的基于批次的ML通常有两个阶段&＃xff1a;模型训练和基于模型的预测。 常用的方法是先使用批处理算法训练模型&＃xff0c;然后将预先训练的ML模型导入流处理应用程序。 只有少数具有内置ML功能的流处理器。 为了将预先训练的ML模型加载到流处理应用程序中&＃xff0c;需要以众所周知的格式&＃xff08;例如PMML&＃xff09;保存模型&＃xff0c;以便可以加载模型而没有任何格式问题。

一些流处理器还支持流机器学习&＃xff0c;该技术可以随着数据的引入来构建和改进模型。但是&＃xff0c;流机器学习的准确性要低于使用经典&＃xff08;基于批处理&＃xff09;机器学习构建的模型。 同时&＃xff0c;经典模型不会随着数据而改善。 因此&＃xff0c;需要不时更新模型。

这是一个权衡。 如果您的流媒体应用程序需要机器学习模型&＃xff0c;则需要考虑精度与概念漂移&＃xff08;模型与新数据之间的漂移速度如何&＃xff09;之间的权衡&＃xff0c;在批处理和流式机器学习之间进行选择。

邮件处理保证

“您希望您的应用展示出多少精度&＃xff1f;” 由流处理器提供的消息处理保证级别决定。 消息处理保证&＃xff08;即语义&＃xff09;确定流处理器的消息传递的可靠性。 您想使用哪种类型的语义作为框架取决于在您的用例中有意义的内容。 最多有一次&＃xff0c;至少一次和恰好一次调用了三种主要的消息处理保证。

最弱的保证最多一次。 实际上最多一次不能保证它将提供任何东西。 例如&＃xff0c;两个事件可能会从传入流进入应用程序&＃xff0c;而只有一个消息可能会从应用程序中发出。 最多执行一次时&＃xff0c;由于不涉及确认协议 &＃xff0c;因此在发送方或接收方不需要任何状态。 一次最多不会带来额外的性能开销&＃xff0c;并且在三个消息处理保证中最容易实现。

一个更强有力的保证是至少一次&＃xff0c;其中消息可能至少被处理一次。 这意味着邮件肯定会发送给收件人&＃xff0c;但是有重复的可能性。 系统故障将导致同一消息在输出流中重复多次。 因此&＃xff0c;下游应用程序需要仔细处理此类重复消息。 如果数据是幂等的&＃xff0c;那么至少使用一个消息处理保证就足以满足您的用例。 为了至少实现一次&＃xff0c;发件人必须跟踪将消息发送给谁&＃xff0c;尚未确认哪些消息以及何时重新发送消息。 对于许多用例&＃xff0c;至少一次消息保证就足够了。 例如&＃xff0c;对于警报生成应用程序&＃xff0c;至少一次消息保证就足够了&＃xff0c;因为它可以确保将警报通知成功传递到预期的参与者。

消息处理保证的第三种类型是一次处理&＃xff0c;这是大多数流处理社区所寻求的&＃xff0c;尽管它可能有意义也可能没有意义。 某些关键任务应用程序&＃xff08;例如&＃xff0c;订单&＃xff0c;金融交易系统&＃xff0c;航班预订&＃xff0c;酒店预订等&＃xff09;仅需要一次担保。如果有两条消息进入该应用程序&＃xff0c;并且发生了某些系统崩溃&＃xff0c;那么一次就可以保证每条消息都得到正确处理一度。 在实现仅一次消息保证时&＃xff0c;我们需要在发送者和接收者上都保持状态。 接收者还必须至少一段时间跟踪过去已经查看和处理过哪些消息。 因此&＃xff0c;在三种消息处理保证中&＃xff0c;这种方法实现起来最昂贵。

许多流处理用例最多只需要一次&＃xff0c;至少要保证一次。 我们观察到&＃xff0c;在19个流处理器中&＃xff0c;有47&＃xff05;的语言至少支持一次语义&＃xff0c;37&＃xff05;的语言仅支持一次语义&＃xff0c;而21&＃xff05;的语言根本没有消息处理保证。

如果您的流应用程序需要更强有力的保证&＃xff0c;则应相应地选择流处理器。

乱序处理和处理不确定性

如果输入数据的时间顺序与源数据产生的时间顺序不同&＃xff0c;将会发生什么&＃xff1f; 流中的无序事件&＃xff08;即&＃xff0c;迟到事件&＃xff09;是由于多种原因而产生的&＃xff0c;例如分布式传感器&＃xff0c;操作员并行化&＃xff0c;网络等待时间以及异步流的合并。 无订单事件处理可能会导致错误的结果。 例如&＃xff0c;需要处理乱序事件以确保事件模式匹配的正确操作。 如果事件X在事件Y之后到达&＃xff0c;则可以配置为生成警报。假设查询检测到事件顺序为Y&＃xff0c;后跟X。如果查询由于异常而在Y之后收到事件X&＃xff0c;则该查询不匹配。 同样&＃xff0c;具有基于事件时间触发的时间窗口的应用程序也可能会受到此类乱序事件的影响。 但是&＃xff0c;某些应用程序&＃xff08;例如事件过滤器&＃xff09;不受乱序事件的影响。

乱序事件处理是一个积极探索的主题。 有几种高级方法。

首先是拦截事件&＃xff0c;然后在处理事件之前将其在缓冲区中重新排序。 此方法增加的延迟与每个事件可以延迟的数量成正比。 当无序引起的延迟较小且有界时&＃xff0c;此方法有效。

第二种方法是构建可处理故障事件并从故障事件中恢复的弹性运算符。 但是&＃xff0c;这使操作员实施起来很复杂。 一个例子是标点符号的使用。 基于标点的技术依赖于随数据流一起发送的特殊元组。 标点符号明确通知查询操作员何时返回Windows结果。 因此&＃xff0c;与上述基于缓冲区的技术不同&＃xff0c;查询运算符可以直接使用乱序输入。

第三种方法称为推测 &＃xff0c;其中当观察到乱序事件时&＃xff0c;我们应用补偿技术来更正先前发出的不准确查询结果。 在这里&＃xff0c;我们假定元组按顺序到达&＃xff0c;并在关闭窗口时立即产生窗口结果。 当检测到延迟到达e时&＃xff0c;受e影响的先前发出的结果将无效。 通过考虑e可以产生这些结果的新修订。 当迟到的人数较少时&＃xff0c;此方法最有效。 如果流连续接收到乱序事件&＃xff0c;这可能会引入额外的延迟以进行更正。

大多数流处理程序不执行无序事件处理&＃xff0c;尽管已经为处理这种疾病的各种方法进行了大量研究。 在接受调查的19家公司中&＃xff0c;只有47&＃xff05;的流处理器关心乱序事件。 此外&＃xff0c;在实际进行重新排序的47&＃xff05;中&＃xff0c;只有少数流处理器。 其他人只是放弃了乱序事件。

性能

具有事件持久性的典型流处理应用程序大约每秒运行5万个事件。 因此&＃xff0c;大多数流处理方案都可以使用2节点HA设置来实现&＃xff08;例如&＃xff0c;请参阅“流处理器是否肥胖&＃xff1f;” &＃xff09;。 但是&＃xff0c;可能需要将系统扩展到2个以上的节点。 此外&＃xff0c;状态大的窗口可能会消耗大量的系统内存。 流处理器需要支持处理如此大的状态。

系统可扩展性

如果您的应用必须处理不断增加的数量负载&＃xff0c;将会发生什么&＃xff1f; 系统可伸缩性和性能是衡量流处理器处理大型工作负载能力的重要指标。 预期的性能水平取决于用例&＃xff0c;一旦选择&＃xff0c;切换到其他流处理器是一项非常昂贵的操作。

例如&＃xff0c;用例最初可能是对小事件的简单过滤操作&＃xff08;很少的字段总计几百个字节&＃xff09;。 但是随着时间的流逝&＃xff0c;事件可能会随着添加具有大量数据&＃xff08;例如&＃xff0c;几兆字节&＃xff09;的新字段而变得更大。 此外&＃xff0c;随着时间的流逝&＃xff0c;业务不断发展&＃xff0c;应用程序逻辑也可能通过复杂的联接&＃xff0c;事件模式匹配&＃xff0c;窗口处理等得到增强。通过这种更改&＃xff0c;用例可能要求的性能要比通过交付实现的性能高出几倍。初始系统。 因此&＃xff0c;如果用例需要很高的性能&＃xff0c;则需要测试候选SP是否可以处理如此繁重的工作负载。

可伸缩流处理器可以通过添加更多资源来扩展其操作规模。 如果保证流处理器从固定的一组输入源&＃xff08;例如&＃xff0c;用于车载网络的流处理器&＃xff09;仅接收固定速率的输入事件&＃xff0c;则该流处理器不需要具有可伸缩性功能。 但是&＃xff0c;如果流处理器例如作为软件服务运行&＃xff0c;需要处理不断变化的工作负载 &＃xff0c;则通常需要扩展能力。 大多数流处理器讨论了它们的水平缩放能力&＃xff08;添加新节点&＃xff09;&＃xff0c;而没有关于垂直缩放&＃xff08;添加资源以保持相同节点&＃xff09;的详细信息。 考虑流处理器的垂直可伸缩性是一个非常重要的因素&＃xff0c;尤其是在无法为部署环境配置更多服务器的情况下。 此外&＃xff0c;水平缩放引入了额外的网络通信开销。 这意味着在进行水平缩放之前&＃xff0c;我们应该首先尝试尽可能垂直缩放。

开源分布式流处理器已发布了每秒数百万个事件范围内的吞吐量数字。 大多数专有流处理器没有公开的吞吐量数字。 给定固定的稳定输入数据流&＃xff0c;流处理器的吞吐量受多个因素的影响&＃xff0c;例如队列大小&＃xff0c;CPU内核&＃xff0c;内存和节点间带宽。 因此&＃xff0c;几乎不可能找到针对所有产品清单进行的研究。 在分布式流处理器上可以找到的性能数字中&＃xff0c;发现的最大吞吐量为1.1亿个事件/秒&＃xff0c;低延迟为30毫秒。 发现的单节点版本的最佳延迟为0.003ms。

处理长聚合窗口

如果应用程序涉及具有数百万个事件的窗口&＃xff0c;将会发生什么&＃xff1f; 通常&＃xff0c;流处理对数据流进行轻量级的一次性处理。 流应用程序通常必须在窗口顶部进行聚合。 窗口的大小很重要&＃xff0c;因为必须将窗口保存在内存中。 如果窗口可能包含数百万个事件&＃xff0c;则必须将其保存在内存中&＃xff0c;并且系统可能会过载&＃xff0c;这将严重威胁系统的稳定性。 例如 &＃xff0c;在电信系统中&＃xff0c;可能存在使用长途电话记录的用例&＃xff0c;这些记录包括每天为1亿客户提供的3亿条记录。 另一个应用程序是网络流量工程&＃xff0c;其中有关当前网络性能&＃xff08;例如&＃xff0c;延迟和带宽&＃xff09;的信息在线汇总&＃xff0c;并用于动态监视和调整网络性能。

可以从流处理器端以及应用程序端提供解决方案。 为了处理较大的窗口&＃xff0c;必须为流处理器提供足够的内存。 如果不是&＃xff0c;则流处理器必须具备通过压缩窗口或将窗口的一部分存储在辅助存储器中来减少内存占用的功能。

从应用程序的角度来看&＃xff0c;解决方案是使用较小的数量进行聚合&＃xff0c;然后基于较小的数量计算较高级别的值&＃xff08;即增量聚合 &＃xff09;。 例如&＃xff0c;如果通过维持3亿条记录的日长窗口预期的最终结果是计算电话的总持续时间&＃xff0c;则可以通过在每分钟明智地汇总通话记录来获得相同的结果&＃xff0c;并使用这些结果来计算小时总和&＃xff0c;最后使用这些小时值来计算一整天的总时长。

结论

不同的流处理器具有固有的功能&＃xff0c;可以针对不同的用例进行匹配。 在尝试为您选择最适合的流处理器时&＃xff0c;要考虑的功能很多&＃xff0c;这使得难以做出正确的选择。

本文介绍了一种选择流处理器的系统方法。

本文通过介绍两个概念来做到这一点。

• 必须具有功能-这些是大多数Streaming应用程序在其生命周期中将需要的功能
• 常规产品—此概念产品包含功能⅔或更多支持的流处理产品。 通过将两个流处理器与普通产品进行比较&＃xff0c;可以决定比较两个流处理器。

该方法回答了两个主要问题。 首先&＃xff0c;流处理器在多大程度上支持核心流处理器架构功能&＃xff1f; 其次&＃xff0c;应用程序的特殊要求是什么&＃xff1f;候选流处理器在多大程度上满足了这些要求&＃xff1f;

在回答第二个问题时&＃xff0c;我们可以使用“一般产品”的概念作为基准。 您可以通过将其与平均产品进行比较来评估特定的流处理。

翻译自: https://hackernoon.com/how-to-choose-a-stream-processor-for-your-app-based-on-advanced-features-8db438811309

演示基于spark的应用流