Spark Streaming:无状态的整体窗口与保持状态

Question

在使用Spark Streaming处理顺序有限事件会话流时,选择无状态滑动窗口操作(例如reduceByKeyAndWindow)与选择保持状态(例如通过updateStateByKey或新mapStateByKey)会有什么考虑因素？

例如,请考虑以下情形:

可穿戴设备跟踪佩戴者进行的身体锻炼.设备会自动检测锻炼开始的时间,并发出消息; 在锻炼期间发出额外的信息(例如心率); 最后,在练习完成后发出消息.

期望的结果是每个运动会话的聚合记录流.即,应该将同一会话的所有事件聚合在一起(例如,以便每个会话可以保存在单个DB行中).请注意,每个会话的长度都是有限的,但来自多个设备的整个流是连续的.为方便起见,我们假设设备为每个锻炼课程生成一个GUID.

我可以看到使用Spark Streaming处理这个用例的两种方法:

1. 使用不重叠的窗口,并保持状态.每个GUID保存一个状态,所有事件都与之匹配.当新事件到达时,状态被更新(例如,使用mapWithState),并且如果事件是"运动结束时",则将发出基于状态的聚合记录,并且移除密钥.

2. 使用重叠的滑动窗口,并仅保留第一个会话.假设长度为2且间隔为1的滑动窗口(参见下图).还假设窗口长度为2 X(最大可能的运动时间).在每个窗口上,事件由GUID进行攻击,例如使用reduceByKeyAndWindow.然后,转储从窗口后半部分开始的所有会话,并释放剩余的会话.这使得每个事件只能使用一次,并确保属于同一会话的所有事件将聚合在一起.

方法#2的图表:

Only sessions starting in the areas marked with \\\ will be emitted. 
-----------
|window 1 |
|\\\\|    |
-----------
     ----------
     |window 2 |
     |\\\\|    |  
     -----------
          ----------
          |window 3 |
          |\\\\|    |
          -----------

我看到的利弊:

方法#1的计算成本较低,但需要保存和管理状态(例如,如果并发会话数增加,则状态可能比内存大).但是,如果最大并发会话数有限,则可能不是问题.

方法#2的成本是两倍(每个事件处理两次),并且具有更高的延迟(2倍最大运动时间),但更简单且易于管理,因为没有保留任何状态.

处理这个用例的最佳方法是 - 这些方法中的任何一种都是"正确的",还是有更好的方法？

应该考虑哪些其他优点/缺点？

Answer 1

通常没有正确的方法,每个方法都有权衡.因此,我将添加额外的方法,并将概述我对其利弊的看法.所以你可以决定哪一个更适合你.

外部状态方法(方法#3)

您可以在外部存储中累积事件的状态.卡桑德拉经常被用于此.您可以单独处理最终和正在进行的事件,例如:

val stream = ...

val ongoingEventsStream = stream.filter(!isFinalEvent)
val finalEventsStream = stream.filter(isFinalEvent)

ongoingEventsStream.foreachRDD { /*accumulate state in casssandra*/ }
finalEventsStream.foreachRDD { /*finalize state in casssandra, move to final destination if needed*/ }

trackStateByKey方法(方法#1.1)

它可能是潜在的最佳解决方案,因为它消除了updateStateByKey的缺点,但考虑到它刚刚作为Spark 1.6版本的一部分发布,它也可能存在风险(因为某些原因它不是很广告).如果您想了解更多信息,可以使用该链接作为起点

优点缺点

方法#1(updateStateByKey)

优点

• 易于理解或解释(对团队的其他成员,新人等)(主观)
• 存储:更好地使用内存只存储最新的运动状态
• 存储:仅保留正在进行的练习,并在完成后立即丢弃
• 延迟仅受每个微批处理的性能限制

缺点

• 存储:如果键的数量(并发练习)很大,则可能不适合您的群集内存
• 处理:它将为状态图中的每个键运行updateState函数,因此如果并发运算的数量很大 - 性能将受到影响

方法#2(窗口)

尽管可以通过Windows实现您所需的功能,但在您的场景中它看起来更不自然.

优点

• 某些情况下的处理(取决于数据)可能比updateStateByKey更有效,因为updateStateByKey倾向于在每个键上运行更新,即使没有实际更新

缺点

• "最大可能的运动时间" - 这听起来像一个巨大的风险 - 它可能是基于人类行为的非常随意的持续时间.有些人可能会忘记"完成运动".也取决于运动的种类,但可能从几秒到几小时,当你想要快速运动的低延迟,同时必须保持潜伏期一样高,可能存在最长的运动
• 感觉更难以向别人解释它将如何起作用(主观)
• 存储:必须将所有数据保留在窗口框架内,而不仅仅是最新的数据.只有当窗口从这个时间段滑落时才会释放内存,而不是在练习完成时.虽然如果你只保留最后两个时间段可能没有太大的区别 - 如果你试图通过更频繁地滑动窗口来获得更大的灵活性,它会增加.

方法#3(外部状态)

优点

• 易于解释等(主观)
• 纯流处理方法,意味着spark负责对每个单独事件起作用,但不试图存储状态等(主观)
• 存储:不受存储状态的集群内存限制 - 可以处理大量的并发运行
• 处理:状态仅在有实际更新时更新(与updateStateByKey不同)
• 延迟类似于updateStateByKey,并且仅受处理每个微批次所需的时间的限制

缺点

• 架构中的额外组件(除非您已经使用Cassandra作为最终输出)
• 处理:默认情况下比仅处理spark更慢,因为内存不需要通过网络传输数据
• 你将不得不实现一次语义输出数据到cassandra(对于foreachRDD期间工人失败的情况)

建议的方法

我试试以下内容:

• 测试数据和群集的updateStateByKey方法
• 看看即使有大量的并发练习也能接受内存消耗和处理(预计在高峰时段)
• 如果没有,我会回到卡桑德拉身边