Kafka+Spark Streaming保证exactly once语义

灰太狼 2023-02-18 14:20 13阅读 0赞

![format_png][]

在Kafka、Flink、Spark Streaming等分布式流处理系统中（Kafka本质上市流处理系统，不单是MQ），存在三种消息传递语义（Message Delivery Semantics）：

*  At Least Once

每条消息会被收到1次或多次。例如发送方S在超时时间内没有收到接收方R的通知，或者收到了R的报错，就会不断重发消息直至R传回ACK

*  At Most Once

每条消息会被收到0次或1次。S只负责向R发送消息，R也没有任何通知机制。无论R最终是否收到，S都不会重发

*  Exactly Once

是上面两个的综合，保证S发送的每一条消息，R都会“不重不漏”地恰好一次收到

一个Spark Streaming程序由三步组成：输入、处理逻辑、输出

要达到Exactly Once的理想状态，需要三步协同进行，而不是只与处理逻辑有关

**Kafka与Spark Streaming集成时有两种方法：**

*  基于receiver
 *  基于direct

### 1、基于receiver ###

![format_png 1][]

*  基于receiver的采用kafka高级消费者API
 *  每个executor进程都会不断拉取消息，并同时保存在executor内存与HDFS上的预写日志（Write-Ahead log，WAL）
 *  当消息写入WAL后，自动更新ZK中的offset

它可以保证At Least Once语义，但无法保证Exactly Once语义。虽然引入了WAL来保证消息不会丢失，但还有可能会出现消息已经写入WAL，但offset更新失败的情况，Kafka就会按上一次的offset重新发送消息。

这种方式还会造成数据冗余（Kafka broker中一份数据，spark executor中一分），是吞吐量和内存利用率降低

### 2、基于direct ###

![format_png 2][]

*  基于direct的方法采用Kafka的简单消费者API，它的流程大大简化。
 *  executor不再从Kafka中连续读取消息，也消除了receiver和WAL。
 *  Kafka分区与RDD分区一一对应，更可控

*  driver线程只需要每次从Kafka获得批次消息的offset range
 *  然后executor进程根据offset range去读取该批次对应的消息
 *  由于offset在kafka中能唯一确定一条消息，且在外部只能被Streaming程序本身感知到，因此消除了不一致性，达到了Exactly Once

不过由于它采用了简单消费者API，我们需要自己来管理offset。否则一旦程序崩溃，整个流只能从earliest或者latest点恢复。

Spark RDD之所以被称为“弹性分布式数据集”，是因为它具有不可变、可分区、可并行计算、容错的特征。一个RDD只能由稳定的数据集生成，或者从其他RDD转换（transform）得来。如果在执行RDD lineage的过程中失败，那么只要元数据不发生变化，无论重新执行多少次lineage，都会得到同样的、确定的结果

Spark Streaming的输出一般是靠**foreachRDD()算子来实现，它默认是at least once的。**如果输出过程中出错，那么就会重复执行知道写入成功。为了让它符合Exactly once，可以施加两种限制之一：**幂等性（idempotent write）**和**事务性写入（transactional write）**

#### 2.1 幂等性写入 ####

幂等性原来是数学里的概念，即f(f(x))=f(x)。

幂等写入就是写入多次与写入一次的结果完全相同，可以自动将at least once转化为Exactly once。这对于自带主键或主键组的业务比较合适（如：各类日志、MySQL binlog），并且实现起来简单

但是它要求处理逻辑是map-only的，也就是只能包含转换、过滤等操作，不能包含shuffle、聚合等操作。如果条件更严格，就只能采用事务性写入方法

stream.foreachRDD { rdd =>
          rdd.foreachPartition { iter =>
            // make sure connection pool is set up on the executor before writing
            SetupJdbc(jdbcDriver, jdbcUrl, jdbcUser, jdbcPassword)
    
            iter.foreach { case (key, msg) =>
              DB.autoCommit { implicit session =>
                // the unique key for idempotency is just the text of the message itself, for example purposes
                sql"insert into idem_data(msg) values (${msg})".update.apply
              }
            }
          }
        }

#### 2.2 事务性写入 ####

这里的事务和DBMS中的事务含义基本相同，就是对数据进行一系列访问与更新操作所组成的逻辑块。为了符合事务性的ACID特性，必须引入一个唯一ID标识当前的处理逻辑，并且将计算结果与该ID一起落盘。

ID可以由主题、分区、时间、offset等共同组成

事务操作可以在foreachRDD()时进行。如果数据吸入失败，或者offset吸入与当前offset range不匹配，那么这一批次数据都将失败并且回滚

// localTx is transactional, if metric update or offset update fails, neither will be committed
        DB.localTx { implicit session =>
          // store metric data
          val metricRows = sql"""
        update txn_data set metric = metric + ${metric}
          where topic = ${osr.topic}
        """.update.apply()
          if (metricRows != 1) {
            throw new Exception("...")
          }
     
          // store offsets
          val offsetRows = sql"""
        update txn_offsets set off = ${osr.untilOffset}
          where topic = ${osr.topic} and part = ${osr.partition} and off = ${osr.fromOffset}
        """.update.apply()
          if (offsetRows != 1) {
            throw new Exception("...")
          }
        }

[format_png]: http://imgconvert.csdnimg.cn/aHR0cDovLzQ3LjExMi4yMjMuOTQvd3AtY29udGVudC91cGxvYWRzLzIwMjAvMDYvJUU1JTlCJUJFJUU3JTg5JTg3LTI3LnBuZw?x-oss-process=image/format,png
[format_png 1]: http://imgconvert.csdnimg.cn/aHR0cDovLzQ3LjExMi4yMjMuOTQvd3AtY29udGVudC91cGxvYWRzLzIwMjAvMDYvJUU1JTlCJUJFJUU3JTg5JTg3LTI4LnBuZw?x-oss-process=image/format,png
[format_png 2]: http://imgconvert.csdnimg.cn/aHR0cDovLzQ3LjExMi4yMjMuOTQvd3AtY29udGVudC91cGxvYWRzLzIwMjAvMDYvJUU1JTlCJUJFJUU3JTg5JTg3LTI5LnBuZw?x-oss-process=image/format,png