标签：窗口 Key Dataflow 流式处理广告 Computation MillWhell 日志

Storm 的不足

Storm 利用异或操作实现了消息至少处理一次；kafka中利用存储在ZooKeeper的offset使得消息队列重放更加容易。Kafka和Storm组合实现了 At Least Once 消息处理机制。但只有实现“正好一次”（Exactly Once）才能得到正确的结果，为此“容错能力”很重要。

实践中使用Storm有哪些问题？
为什么 Exactly Once 消息处理机制难点在哪？为什么必须？
Storm的容错机制缺失了什么？
“时间窗口”（Time Window）是什么？为什么重要？

一个简单的流式数据处理系统

对于一个广告点击率计算和计费的数据处理需求，日志格式如下：

前3个ID表示了广告的位置
事件类型表示这条日志是一次广告展示还是点击
UID指定用户，可对于一个用户短时间点击多个想通过广告去重
事件ID，标识唯一事件。在实现“正好一次”时，使用它进行去重
花费字段，花费了广告客户多少预算

实际中复杂的多，可能有上百个字段，如IP、地理位置等。有了这个简单的日志格式后，可做两个最常见广告数据的流式处理：

接近实时的广告计费。实时计算广告客户消费，保证不超过设置的预算
统计各广告点击率

使用Storm和Kafka搭建系统，应用服务器将日志发给负载均衡，再均匀地发给kafka集群的Broker，下游的Storm集群有一个Topology完成广告计费和广告的点击率统计。

Topology 中KafkaSpout从kafka拉去日志发给下游Bolt。发送数据采用字段分组方式

AdsCtrBolt 计算广告点击率，接收广告ID分组
内存维护广告ID=>（展示次数，点击次数，广告花费）的Map，定时输出到外部存储，如HBase，即每分钟输出一次对应广告ID点击率
ClientSpentBolt 计算广告客户的花费，接收广告客户ID分组
以更高频率，如每秒或每次接收广告点击，就对应更新一次HBase里广告花费数据

Storm 的 Topology 开启了 AckerBolt，能保证所有消息至少被处理一次

“正好一次”的正确性

但在大数据和分布式情况下，始终面临“出错”。“至少一次”已无法满足业务需要，如广告计费要求很高的准确性。如某个ClientSpentBlot写入外部存储时出现高延迟，这时“至少一次”的处理机制会重发消息，ClientSpentBolt会重复计算同一条日志的广告花费。

如果某个KafkaSpout挂掉了，那可能有一大批消息要重复计费了。因为考虑到性能，从Kafka拉去数据不是拉一条、处理一条、更一次ZooKeeper的offset。尤其是ZooKeeper是处理都的分布式锁，不能承受大负载。KafkaSpout从Kafka拉一小批数据再发送，等处理完了再更新一次ZooKeeper上的偏移量。只要有一条消息没处理完，KafkaSpout挂掉了且offset没更新。容错机制会再启动一个KafkaSpout并重新拉去一次这批数据。这时就会对一大批日志重新计费。

一个直观的解决方法是对重复消息去重，在每个Bolt中维护其已处理完的所有message-id的集合。每收到新消息就到集合看看是否已经处理过。每个Bolt保留所有处理过的message-id占用太多内存，可进行优化：

BloomFilter 代替数据集合
副作用是有很小概率将不重复也认为是重复的
将数据按照时间窗口，切分成多个BloomFilter
如设定30个布隆过滤器，每个只存放一分钟的message-id，每过一分钟清空一个最早的。通过一个固定大小内存确保30分钟内重复数据不会被多次处理。时间窗口可设置

“正好一次”是现代流式数据处理第一个目标

计算节点迁移的容错问题

BloomFilter 引入了一个新的“状态”，而且数据处理需求本身也有状态。如AdsCtrBolt中广告ID=>（展示次数，点击次数，广告花费）的Map就是Bolt中维护的状态。维护状态带来新的问题，系统的容错问题。

节点的容错很容易，只要在其他机器重启一台即可。但状态的容错较困难。此外系统的可扩展性也要考虑Blot中的状态，Storm论文中并行度是部署Topology时预先设定的，很难动态扩容。提高并行度意味着：

在线增加服务器数量，运行中的Bolt会被迁移（Migrate）到其他服务器
增加Bolt数量。这意味着Bolt中Map状态也要能拆分，这时S4中每个PE对应一个Key的设计更加合理。那样迁移状态的状态和对应计算函数绑定在一起

Bolt 在拆分和迁移时要能够保留状态信息，这意味着状态要能持久化。节点挂掉后，其他节点恢复计算能力时要把这些状态信息重新读取回来。这使得调度计算更加容易，可以动态地增加系统的并行度。

通过计算节点中间状态持久化，使得系统在容错情况下仍能做到“正好一次”，并能线上动态扩容、调度计算，这是现代流式数据处理的第二个目标。

处理消息的时间窗口

AdsCtrBolt 每分钟输出一次广告点击率，这个“每分钟”依靠Storm内建的TickTuple机制。Storm按照指定的时间间隔向每个Bolt和Spout发送一个特殊TickTuple。每当Bolt接收到TickTuple时将当前计算出的状态信息输出出去。

这有个问题：使用消息到达AdsCtrBolt的时间代替了对应广告曝光和点击时间。也就是使用处理时间代替了事件时间。这样计算的结果与真实结果有差异

业务需求上。如广告客户设置广告预算在11月花完，11月30日晚11点59分59秒的广告点击可能被分配到12月1日。这样广告客户看到在12月没有分配广告预算但也有花费
关于日志的重放。通过重放Kafka日志重新计算数据时，会将重放的日志都算在重放日志的极短时间内，这造成巨大错误

时间发生到被处理存在消息传递延时，如图中E1比E2发生的早，但是迟好几分钟才被处理。如果使用事件发生时间，存在新问题：

维护的Map映射要改为三维多层映射 时间窗口=>[广告ID=>(展示次数，点击次数，广告花费)，....]
很难据欸的那个什么时候写入外部存储。上游发送的日志不是严格按时间排序。这要加上几个判断条件：
1. Bolt内部要维护一个“时钟”，判断最近接收日志大概在什么时间戳附近。如最新N条日志的最早时间戳
2. 若时间窗口比最近时间戳晚了一个特定时间长度，就认为不会再接收到这种日志，就可以把对应数据写入外部存储
3. 又有一条确认不再该收到的日志传输过来，可直接忽略或更新到外部数据库

现有的Storm无法做到，像维护时间窗口的映射关系、统计最近日志时间戳等逻辑代码仍需自己撰写。

现代流式计算的第三个目标：把和时间窗口相关的机制、触发数据更新到外部存储的机制，在流式框架中内建，无需开发者关心容错等问题。

小结

三个目标：

“正好一次”的数据处理机制
计算节点要使用的“状态”信息持久化，容错能力
流式数据处理的时间窗口、触发机制内置到流式处理系统中

MillWhell

MillWhell 架构如何，抽象模型和S4、Storm有哪些异同？
除简单中间数据持久化，流式数据处理系统还要考虑哪些容错场景？MillWhell 如何解决？

MillWheel: Fault-Tolerant Stream Processing at Internet Scale

MillWhell 也使用有向无环图DAG，有一下几个组成概念：

流（Stream）
键（Key）

计算（Computation）
对应Storm中Bolt或S4里PE，一个例子定义如下：

computation SpikeDetector {
    input_streams {
        stream model_updates {
            key_extractor ='SearchQuery'
        }
        stream window_counts {
            key_extractor 'SearchQuery'
        }
    }
    output_streams {
        stream anomalies {
            record_format 'AnomalyMessage'
        }
    }

包含3部分

订阅了哪些流，消息输入的流向是什么
输出哪些流，消息输出流向是什么
本身计算逻辑

键（Key）

MillWhell中每个消息可被解析成（Key， Value，TimeStamp）这样的三元组。一个Computation可针对输入的消息流，定义自己的 key_extractor。Storm和S4中消息根据字段进行不同维度划分，发给不同PE或Bolt。在抽象层，这是发送了两个不同消息流。在MillWhell则是一个消息流被两个不同Computation订阅，只是两个Computation可以有不同 key_extractor。这样系统逻辑层可复用同一个流。

Key也是MillWhell中进行计算的唯一单元。也就是一个Computation实现里获取到的都是同一个Key的状态。如广告点击率统计的例子，每个广告ID就是一个Key，一个Computation里获取到的日志记录都是这个某一个广告ID的。

这类似S4的PE，Computation + Key 就是一个 PE。而 Computation + Key 的组合可在不同机器间调度，有助于负载均衡和扩容。但MillWhell没有把一个 Computation+Key 当做一个对象处理，每个Computation里的Key就像Bigtable里的Tablet一样，分成一段一段。实际的负载调度时，调度的也是一整段的Key。这个实现避免了S4对应PE对象过多。Computation+一段Key的组合类似Storm中的一个Bolt，要处理一段Key。

低水位（Low Watermarks）和定时器（Timers）

流式计算框架存在事件实际发生时间和接收到数据时间存在差异的问题。为了让每个Computation进程知道某个时间点之前的日志应该已经处理完了，引入低水位和Injector模块。解决事件差依靠消息三元组中的时间戳。

Computation 处理完一个消息后向下游发送时，也要为新消息创建一个时间戳。新的时间戳不能早于被处理的消息的时间戳。若希望后续处理都基于最早发生的事件，最好直接复制输入消息的时间戳。

低水位是指在Computation中拿到所有未处理完的消息里最早的那个时间戳。此处的“未处理完”包括：还在消息管道待传输的消息；已经在Computation里存储的消息；处理完但还未向下游发送的消息。一个Computation可能还订阅了多个上游Computation，它们也有同样的时间戳，“低水位”就是一个Computation和它上游的低水位中时间戳最早的那个。

min(oldest work of A, low watermark of C:C outputs to A)

当Computation知道自己的低水位，就能决策应该进一步等待更多消息，还是当前数据已是完整的可以输出。

MillWheel这么做的：每个Computation进程统计自己处理的所有日志的“低水位”信息，上报给一个Injector模块。这个Injector模块收集所有类型的Computation进程所有低水位信息。通过Injector将相应的水位信息发给各个Computation进程，各个Computation自行计算自己的低水位。

每种类型的Computation都有自己的水位信息。同一Computation的不同进程的水位信息不同，因为其处理消息的进度可能不同。不同类型的Computation的水位信息是不同的，因为整个数据流的拓扑图可能会很深。可能前几层处理到10点05分，后几层才处理到10点01分。如果整个拓扑图使用同一个水位信息，意味着前几层统计结果的输出的延时会变大。有了水位信息，统计某时间段数据可做到基本准确。

论文中只有0.001%的日志在考虑了水位后仍会因来的太晚而被丢掉。但因为所有数据都是持久化的，即使消息来的太晚，仍能纠正之前的数据。

MillWheel 提供了一组定时器（Timer）API。根据日志时间戳的得到日志对应时间窗口，可更新数据。再更加时间窗口设置对应Timer，系统根据水位信息触发Timer执行，Timer执行时将对应统计结果输出

// Upon receipt of a record,update the running
// total for its timestamp bucket,and set a
// timer to fire when we have received all
// of the data for that bucket.
void Windower::ProcessRecord(Record input){       // 处理消息，并关联给Timer
    WindowState state(MutablePersistentState());
    state.UpdateBucketCount(input.timestamp());   // 更新统计数据，被持久化
    string id = WindowID(input.timestamp());
    SetTimer(id,WindowBoundary(input.timestamp()));
}
// Once we have all of the data for a given
// window,produce the window.
void Windower::ProcessTimer(Timer timer){          // 根据水位信息在合适时触发，然后对应计算统计信息并输出
    Record record = WindowCount(timer.tag(),MutablePersistentState());
    record.SetTimestamp(timer.timestamp());
    // DipDetector subscribes to this stream.
    ProduceRecord(record,"windows");               // 输出结果，被持久化
}
// Given a bucket count,compare it to the
// expected traffic,and emit a Dip event
// if we have high enough confidence.
void DipDetector::ProcessRecord(Record input){
    DipState state(MutablePersistentState());
    int prediction = state.GetPrediction(input.timestamp());
    int actual = GetBucketCount(input.data());
    state.UpdateConfidence(prediction,actual);
    if (state.confidence()>kConfidenceThreshold){
        Record record = Dip(key(),state.confidence());
        record.SetTimestamp(input.timestamp());
        ProduceRecord(record,"dip-stream");
    }
}

Strong Production 和状态持久化

无论是中间阶段还未向下发送的数据，还是确定向下发送的计算结果，都要持久化。这是为了容错和迁移某段 Computation+Key 到其他服务器。

MillWhell 封装了整个持久化API，可直接通过它进行读写，无需直接与外部存储交互。

每个 Computation+Key 接收消息并处理的过程：

消息去重，可通过分段的 BloomFilter 解决
执行用户实现的业务逻辑代码，所有产生的更新，无论是Timers、State、Production，都被视为对“状态”的变更
状态变更被一次性提交给后端存储，如Bigtable或Spanner
更新持久化后，系统发送Acked消息给上游发消息的Computation。这个Acked用于重发以处理丢失，配合第一步消息去重，实现“正好一次”的数据处理。
不同点：MillWhell中消息会被持久化，所以无需等消息在DAG里处理完再从起点清理，每层可单独回收下一层已处理完的消息
向下游发送消息

第五步发送前向Bigtable/Spanner里持久化消息，持久的内容被称为检查点（Checkpoint），正因为这一步MillWheel才有容错能力和在线迁移计算能力。考虑性能，多个记录放在一个Checkpoint。Checkpoint 类似数据库里的预写日志（WAL），即使节点挂了或要在线迁移节点，只要再其他节点将这个Checkpoint读出再向下游发送。

第三步已经记录了，为什么还要Checkpoint呢？其他节点重启计算进程时直接从中间结果读取不就可以了么？

一个例子：

一个Computation按数据处理时间和时间窗口统计数据并下发。如每5分钟发送自己接收的日志到下游
每来一条日志都按前三步更新统计数据。Timer 根据实际时间（Wall Clock）而不考虑日志里时间戳或水位，每5分钟触发一次，发送统计结果到下游
某个Timer触发时，生成计算结果并发送，此时节点挂掉。消息发送出去但不知下游是否收到。这是第一条消息X
在另一个节点启动这个Computation，但此时一条新日志进来。这是还在同一秒钟，还在同一窗口，会更新对应数据
向下游发送一个新的不同的计算结果，这是第二条消息Y
由于网络传输是乱序的，但不知X和Y哪个先到。X先Y后，Y会因为X被去重，那么下游实际计算中就丢了一条记录

为此，MillWhell 采用了简单的方式：将要向下游发送的数据作为Checkpoint写下。之后才是简单重放Checkpoint日志，避免这样基于时间点的隐式依赖，导致不能做到数据层的一致性。

这个Checkpint策略被称为 Strong Production。MillWhell的数据处理虽支持乱序，但所有的输入数据是严格不重复、也不会丢弃的。

僵尸进程和租约问题

MillWheel有个中心化Master集群进行负载均衡，在节点挂掉时也是由这个Master启动新的Computation进程。但Master判断为节点挂掉，并不意味节点进程真的挂了，可能是网络分区造成的。可能存在两个Computation进程管理同一段Key，旧的Computation进程就是没杀掉的僵尸进程。虽然上游数据被Master调度向新的Computation发送，但旧的Computation可能定时触发Timer，向下游或持久层写数据，导致数据不一致问题。

解决方案就是租约，每次写入都带上租约的Token。MillWhell启动一个新进程进行容错处理时，老进程的租约作废。

所有分布式系统都有类似机制，确保任何一个Key只有写入者，即 Single-Writer 机制

Weak Production 和幂等计算

有些任务无需消息去重和Checkpoint等有大量开销的机制。这种 Computation 是无状态的，没有中间计算结果，只要简单地让上游重发即可，无需持久化。关闭 Strong Production 后地 Computation 节点被称为 Weak Production

小结

MillWhell 接管了数据存储层，中间结果和输出都存入Bigtable或Spanner等
数据去重：每个收发的消息都创建了唯一id，在每个Computation的每个Key上都通过Bloomfilter对处理过的消息去重，确保操作幂等
流式计算的容错和扩容：通过Strong Production 方式，对所有向下游发送数据创Checkpoint。类似数据库的WAL。
事件创建时间和处理时间间的差异：MillWhell 引入一个独立的 Injector 模块，其收集所有计算节点进度，也反馈给各节点最新“低水位”。这样可以在数据窗口对应数据处理完后再输出准确结果
流式计算的容错问题：避免僵尸进程向持久层写数据，这通过向每个工作进程注册一个id确保一个Key只有一个写入者。通过租约实现，与GFS Bigtable 类似

MillWhell 解决了数据正确性、系统容错能力、数据处理的时间窗口。

缺点：

没有考虑“流批一体”
对于时间窗口，从实际应用层面进行设计，但没有总结抽象一个模型

Dataflow

书籍 Streaming Systems

各种流式处理系统都使用了DAG，但具体实现和接口又都不相同。
S4无中心，一切皆PE；
Storm是中心化架构，定义了发送数据的Spout和处理数据的Bolt；
MillWhell定义了 Computation、Stream、Key 等DAG的概念，还引入了 Timer、State 等为了持久化状态和处理时钟差异的概念

以上都是具体的数据处理系统，而非高度抽象的编程模型。都是从具体实现的“是怎么样”角度触发，没有从模型角度“该怎么样”抽象出来。但相识之处：都采用DAG，将同一个Key的数据再逻辑上作为一个单元抽象。

从抽象角度理解流式数据处理
使用S4、Storm、MillWheel 拓展 Dataflow 抽象模型，实现概念的落地

Dataflow 基础模型

两个概念：ParDo，并行处理；GroupByKey，按照Key进行分组数据处理。

ParDo，类似MapReduce里的Map。输入数据被DoFn处理函数处理，数据不是在一台服务器上，而是和MapReduce一样在很多台机器上并行处理。只是Map和Reduce都只有一个，Pardo会和GroupByKey组合成很多层，像多个MapReduce串在一起

GroupByKey，类似MapReduce的Shuffle操作。Dataflow中所有数据被抽象为 KV 对，ParDo和Map的输入输出都是kv对。GroupByKey将相同的key汇总，再通过下一个ParDo下的DoFn进行处理

例如统计所有广告展示次数超过100万次的，可通过Pardo解析日志，然后输出（广告ID，1）这种KV对，通过GroupByKey将相同ID的数据分组合并。再通过一个ParDO并行统计每个广告ID下展示次数。再通过一个ParDo过滤掉所有少于100万次展示的广告

流批一体

Dataflow与多个MapReduce的组合最大不同就是时间维度。GroupByKey将相同Key的数据Shuffle到一起供后续使用，但没定义Shuffle的时间。

MapReduce模型下输入数据再任务开始前就定义好了，Dataflow中输入的数据集是无边界的，随时间推移不断加入新数据。对一份预先定义、边界明确的数据同样可使用流式处理。对于不断增长的实时数据，也可不断执行MapReduce等批处理任务或 Spark Streaming 等微批（Mini-Batch）的处理方式

流式计算也会通过微批的方式提升性能，如MillWhell中的Checkpoint就是等待多条记录处理完成后批量进行。

Kappa架构实现了流批一体，而Dataflow提出批处理是流处理的特殊情况

时间窗口的分配与合并

MillWheel 中流式数据处理系统已经很完善了，但对“时间”的处理还很粗糙。MillWheel开始区分事件处理时间（Processing Time）和事件发生事件（Event Time），也引入时间窗口概念。对于计算结果何时输出，仍采用简单的定时器（Timer）方案。Dataflow主要对这些该概念进行抽象。

流式处理中，往往使用时间窗口，常用时间窗口分成：

固定窗口（Fixed Window）如“每小时广告展示数量”
滑动窗口（Sliding Window）如“过去2分钟广告展示量”，划分为 12:00~12:02、12:01~12:03 这种，窗口大小为2分钟，滑动周期为1分钟
会话窗口（Session Window）常用于统计用户会话，往往通过设置两次事件间“超时时间”定义会话。如，一个客服聊天系统，若30分钟没有互动就认为会话结束，再有新发言则认为进入新会话。

Dataflow模型在统计数据时往往需要 GroupByKeyAndWindow，需要根据特定的时间窗口进行数据统计。最重要的函数：AssignWindows 、MergeWindows。

在业务处理函数前事件都是（key，value，event_time），AssignWindows 就是将其根据处理逻辑变成（key，value，event_time，window）。一个事件可能分配给多个相邻的时间窗口，此时一条记录变成多条记录。可以利用 Key+Window 得到固定窗口或滑动窗口统计数据。

例如第三种会话窗口等就很难定义。Dataflow 里通过 AssignWindows+MergeWindows 组合进行数据统计。以客服30分钟超时为例：

根据同一用户行为分析，Key为用户ID，对于Value为用户或客服发的消息，event_time 时实际发送时间
每个事件进行AssignWindows时，将对应时间窗口设为 [eventtime, eventtime+30)，也就是事件发生后的30分钟内都是这个事件对应会话的时间窗口
同一个Key的多个事件对应的窗口合并，对于会话窗口，如果两个时间窗口有重合就合并成一个更大窗口（MergeWindows），所有事件合并合并完成后几个时间窗口就对应几个会话

窗口的分配和合并使得Dataflow可处理乱序数据。乱序到达的数据计算结果都是相同的。上次计算完的结果作为状态持久化，进入新事件后按照 AssignWindows 和 MergeWindows 不断进行数据化简

触发器和增量数据处理

有了窗口函数逻辑就可统计会话数量，乱序数据也可处理。实际中输入数据以流的形式传输。且可能遇到延时、容错等情况，所以还要一个机制告诉什么时候数据到了，可以把计算结果向下游输出。

MillWheel中通过低水位（Low Watermark）判断是否该处理的事件处理完了，可以向下游发结果。但存在问题：

实际的水位标记后，仍有新日志到达。这时向下游发送的数据不准确，对于广告计费等高精度要求的业务难以接受
因为考虑所有节点，只要一条日志来晚了水位就特别低，导致迟迟无法输出计算结果。

通过Lambda架构解决，不搭建一个批处理层，而是尽快给出一个结果，在后续根据获得的新数据不断修正计算结果。在Dataflow中体现为触发器（Trigger）。MillWheel里只能通过定时器向下游发送数据，本质上只有“时间”一个维度，可以根据当前水位和时间判断是否触发。Dataflow中除了基于水位的完成度触发器，还支持基于处理时间、记录数等多个参数组合触发。用户可自定义触发器

// Apache Beam 示例
PCollection<String>pc=···;
pc.apply(
    Window.<String>into(
        FixedWindows.of(1,TimeUnit.MINUTES) // 设立一分钟的固定窗口
    ) 
    .triggering(
        // 第一条数据被处理后，延迟一分钟触发
        AfterProcessingTime.pastFirstElementInPane().plusDelayof(Duration.standardMinutes(1))
    )
    .discardingFiredPanes()
);

确定数据计算的触发后，还可以定义触发后的输出策略：

抛弃策略（Discarding）
触发后窗口内数据抛弃，后续再有窗口内数据到达也无法合并。这样不会占用太大存储空间，适用监控系统统计错误并警告等场景
累积策略（Accumulating）
触发后积累的窗口内数据仍持久化为状态保存。再有新日志来会再计算结果并向下游发送，下游也会用新结果覆盖老结果。典型的Lambda架构，一半采用此策略。计算时延小，且能修正数据。
累积并撤回策略（Accumulating & Retracting）
除了“修正”结果还要“撤回”计算结果。如存在两个会话，再来一条日志使得两个旧会话加它形成一个新会话。此时不仅要发送新会话，还要撤回旧会话（Apache Beam 仍没实现）

小结

Dataflow 中，大数据流式处理抽象成3个概念：

对乱序数据，按照事件发生时间计算时间窗口模型
根据数据处理的多维度特征，决定计算结果什么时候输出的触发器模型
能把数据更新和撤回，与前面的窗口模型和触发器模型集成的增量处理策略

与 MapReduce 类似，Dataflow 不是具体实现，而是从模型角度思考无界大数据处理该如何抽象。

标签：窗口,Key,Dataflow,流式,处理,广告,Computation,MillWhell,日志
From： https://www.cnblogs.com/zhh567/p/17302829.html

大数据经典论文解读 - 流式计算 - MillWhell - Dataflow