Iceberg事务特性解读

Posted 2022-12-04 咬定青松

本文来源微信公众号：码上观世界

一、常规大数据处理技术的痛点

常规的大数据处理技术栈是基于Hadoop、Hive、Spark等实现大数据的抽取、转换和加载（ETL或者ELT），然后基于转化、装载后的数据通过Hive、Impala、Presto等分析和查询等。常规的ETL过程周期通常较长，比如以天为单位，但是随着实时应用需求场景的增多，反向导致ETL的周期越来越短、并发读写的任务也越来越多，从而给常规的ETL过程带来更多的挑战，具体表现在：

1 数据抽取的频率从原来的天缩短到小时、分钟，要求处理技术升级和数据存储能力升级；

2. 面向OLAP的存储系统在实时大数据场景下也要求具备OLTP系统类似的快速存取能力，包括快速写入、更新和查询能力；

3. 分布式计算引擎和分析引擎的引入要求上下游系统能够并发处理，包括并发读和并发写；

4. 频繁的数据源Schema变更（如字段的增、删、改），要求下游数据表能自适应源端的更改；

5. 下游数据源分区变更要求不对历史数据进行重分布；

6. 下游实时数据应用需求要求下游数据支持增量消费，比如批读、流读；

7. 机器学习或者数据挖掘等场景要求下游存储系统支持全量扫描或者指定范围扫描；

8. 在存算分离场景，同样一份数据要能够支持不同计算引擎的共享复用；

9 某些场景下，可能需要数据回滚，这要求存储系统具备一定的版本管理能力；

来看看基于Hadoop、Hive、Spark等技术框架，能否实现上述要求？也许你可以选择通过离线脚本任务，每小时将增量数据加载到Hive表的小时分区，为了进一步提高这个频率，你还可以选择Hive Streaming或者Spark Streaming 将分钟级别的增量数据写入到表的分钟分区中，这固然能够在某种程度上降低数据处理的时延，但是带来以下问题：

a. 过高的频率导致文件和分区数目的暴增，将极大降低分布式系统的响应能力和增加系统的负载，特别是NameNode节点；

b. 无法较好的处理并发写入，比如容易造成写分区冲突，导致任务失败和脏数据；

c. 分区写入过程无法隔离其他引擎的读取，容易造成不完整或者错误的数据读取；

d. 不支持数据的更新、去重，需要额外的作业去合并和去重数据；

e. 当上游数据Schema发生变更，下游表需要重建，特别是当分区变更，需要重写所有数据，该过程非常漫长，对并发读写侵害较大；

f. 当对历史数据分析时，不方便按照时间范围进行数据回溯；

g. 当修正数据时，无版本机制导致无法回滚到指定版本;

二、数据库系统的并发控制与事务

基于Hadoop、Hive、Spark等的技术框架之所以不能很好地实现上述功能，一个很重要的原因是因为并发的存在，但是又没有像数据库那样进行并发控制的事务功能。

在操作系统中，并发是指一个很短的时间段中有几个程序都处于已启动运行状态，到运行完毕状态之间，多个程序之间有相互制约关系，直接制约体现为一个程序需要另一个程序的计算结果，间接制约体现为多个程序竞争同一资源，如缓冲区、数据等。在数据库系统中，并发主要是指资源的争用，当两个进程在同一时刻访问或更新同一个数据时，就会产生资源的争用，当对资源争用不加以控制时，它会引起一系列的问题，比如数据不一致、查询阻塞、死锁等。

SQL Server必须对资源争用加以控制，实现并发控制读操作的不同方法，称作并发模式。当发生读写冲突时，并发模式控制数据处理的顺序，以保证数据的一致性。

事务是所有数据库系统的一个基本概念。一次事务的要点就是它把多个步骤捆绑成了一个单一的，不成功则成仁的操作。其它并发的事务是看不到在这些步骤之间的中间状态的，并且如果发生了一些问题， 导致该事务无法完成，那么所有这些步骤都完全不会影响数据库。事务具有4个特性，统称ACID属性：

• 原子性：事务整体是一个工作单元，对数据的修改操作，要么全部执行，要么完全不执行，没有第三种状态。

• 一致性：在一个事务执行之前和执行之后数据库都必须处于逻辑上的一致性状态，数据在不同的事务中是相同的。

• 隔离性：并发执行的事务之间是相互隔离的，一个事务内部的状态，对其他事务是不可见的。

• 持久性：当系统发生故障时，持久性确保已提交事务的更新不会丢失，也就是说一旦一个事务提交，DBMS保证数据的改变是永久性的，持久性通过事务日志来保证。

ACID 4个属性的关系可大概表述为：原子性是要求，一致性是目标，隔离性是手段，持久性是结果。所以如何做到原子性地隔离是实现事务的重中之重。在宏观上，实现事务特性是通过并发控制。在微观上，实现事务要靠隔离。

SQL Server支持悲观并发模式和乐观并发模式来控制并发，保证事务的ACID属性，两种并发模式的不同之处是保护数据的结构不同。在一个数据库中，只能选择其中一个并发模式，默认的并发模式是悲观并发模式。

1，悲观并发模式

悲观并发模式使用锁结构来保护数据，保证数据的一致性，适用于锁消耗低于回滚事务的成本环境中。

悲观并发模式是默认的并发模式，在数据库系统中，当多个进程访问同一资源时，SQL Server会通过各种类型的锁来协调资源的访问，确保在并发环境下数据保持一致的状态。而锁的作用范围是在事务中，事务建立在并发模式下。

在悲观并发模式下，SQL Server认为有大量的写操作发生，并且写操作会受到写操作的影响。也就是说，悲观并发模式对任何正在访问的数据进行加锁，以避免多个进程同时修改或读取数据，造成数据的不一致。在默认的隔离级别下，读和写是相互阻塞的。

2，乐观并发模式

乐观并发模式在并发过程中不产生锁，而是使用行版本（Row Version）来保护数据，保证数据的一致性，适用于写操作较少的环境中。

在乐观并发模式下，当一个事务对数据进行修改时，SQL Server把已经提交的老版本数据存储在tempdb的版本库（version store）中。读操作读取的是已经提交的老版本，这些数据实际上存储在tempdb中。

对于乐观并发模式，SQL Server假设只有少量的写写冲突发生，默认的机制是使用快照技术，在写进程完成数据修改之前，先把数据的行版本保存到tempdb中。由于数据的老版本已经保存，读进程可以直接读取已经保存的行版本，而不会受到写进程的影响。

乐观并发使得读写操作不会相互阻塞，但是，这会导致一个潜在的问题，读进程可能会读取到老的数据。

每一个事务都运行在一个特定的隔离级别内，该隔离级别是由会话（session）决定的。SQL Server定义的五个隔离级别，都是为了定义读数据的行为：

• READ UNCOMMITTED ：允许脏读、数据不可重复读和数据范围不可重复读

• READ COMMITTED：防止脏读（读取未提交的数据更新），允许数据不可重复读和数据范围不可重复读

• REPEATABLE READ：防止脏读和数据不可重复读，允许数据范围不可重复读

• SERIALIZABLE：防止脏读、数据不可重复读和数据范围不可重复读

• SNAPSHOT：快照隔离级别，主要是为了避免读写相互阻塞

隔离级别定义事务处理数据读取操作的隔离程度，在悲观并发模式下，隔离级别只会影响读操作申请的共享锁（Shared Lock），而不会影响写操作申请的互斥锁（Exclusive Lock），在乐观并发模式下，从tempdb中读取事务开始时已经提交的行版本（row version），而不会读取到尚未提交的数据行版本。

在SQL Server定义的五个隔离级别中Snapshot隔离级别属于乐观并发模式，其他隔离级别属于悲观并发模式。在SNAPSHOT隔离级别下，任何写操作都会将更新之前的数据行保存到tempdb中，读取操作要么从Original Database的数据表中读取数据，要么从tempdb中读取行版本数据。Snapshot隔离级别指定：在一个事务中，任何语句读取的数据，是事务一致性的版本。事务一致性是指在事务开始时，在表级别创建数据快照，只能识别其他事务已提交的数据更新。在事务开始之后，当前事务不会识别其他事务执行的数据更新。如果一个Snapshot 事务尝试去提交数据行的更新，但是该数据行已经被其他事务修改，并且修改的时间早于当前事务开始的时间，那么SQL Server将当前事务作为失败者，并回滚其事务操作。Sanpshot隔离级别实现事务级别的数据一致性。SQL Server 使用tempdb来存储行版本化（row versioning）的数据，如果数据更新较多，存储的行版本太多，会导致tempdb成为系统瓶颈。乐观并发模式用于冲突较少的环境中，如果应用程序在更新数据时经常发生冲突，Snapshot隔离级别可能不是最好的选择。

三、大数据技术中基于快照的事务实现

在大数据场景下，高并发虽然存在，但是被限制到了较低层次，因为大量的写操作被作为事件流批量写入或追加到存储系统，这个过程可以并行无冲突进行，只有在涉及到存储系统元数据更新时才涉及到并发控制。因此，大数据存储系统更倾向使用乐观并发控制而不是悲观并发控制，而Snapshot机制就是实现乐观并发控制的事务特性的直接选项。但是Iceberg的此Snapshot跟数据库的彼Snapshot略有不同：

1. SQL Server 数据库的快照是数据修改前生成，Iceberg的快照是数据修改后生成；

2. SQL Server 数据库的快照保存了只涉及到修改的那部分原始版本数据，Iceberg的快照保存了修改后的最新数据；

比如在Iceberg中创建了两个快照，在第一个快照snap-1中增加了一条记录（1，a），在第二个快照snap-2中新增了一条记录（2，b），两个快照可以并行执行，分别存储到文件df-1.parque和df-2.parquet。在实际使用中，一个快照可能会涉及到多个数据文件，因此，还需要一个数据清单文件存储所有的数据文件信息，数据清单文件本身也可能分不同的类型，比如追加清单文件和删除清单文件，最后用snap指向这些清单文件：

在Iceberg的一个快照内部，实际上拆分为两个步骤来实现的：数据文件的写和数据文件的元数据提交，如下图所示：

数据文件的写和数据文件的元数据提交分别通过Flink的两个算子IcebergStreamWriter和IcebergFilesCommitter来实现，IcebergStreamWriter借助Flink DataStream的并发机制，将数据文件的写操作并行化，可互不影响地写数据，IcebergFilesCommitter 收集IcebergStreamWriter创建的数据文件，来创建manifest数据清单文件和快照的提交。因为提交文件元数据涉及到资源竞争，因此在提交元数据端使用了并发控制：Iceberg根据Catalog存储类型的不同，而使用不同的控制方式，比如使用Hadoop Catalog时，Iceberg基于文件rename来保证元数据的提交不冲突，使用Hive Catalog或者 JDBC Catalog则使用数据库的锁或者MVCC来保证元数据的提交不冲突。那如何保证数据文件的写和数据文件元数据提交这两个阶段的原子性呢？这就依赖Flink的Checkpoint两阶段提交机制了，一旦Flink Checkpoint成功完成，数据文件和元数据都都外可见，否则快照失效。因为快照本身是不变的，所有基于快照的查询看到的结果都一样，因此保证了事务的ACID属性。

Iceberg快照保证了事务的ACID属性，解决了并发的读写问题，同时因为快照本身记录了版本相关的元信息，比如快照的开始和结束时间、数据大小、行数、以及其他一些统计指标，因此Iceberg 快照具备了超过数据库快照更多的能力，尤其是在基于CBO的查询优化。另外，快照还记录了Schema变更-包括分区变更的信息，而不用对原数据重写，提高了上述变更操作的及时性。但是基于这种快照机制的事务也存在一些问题，比如：

a.依赖Flink的Checkpoint时间间隔，如果间隔过大，数据可见性延迟变大，如果时间间隔过小，不仅会产生大量小文件，对性能产生很大影响，而且由于写数据延迟和Checkpoint开销，从而加大Flink的负担。

b. Iceberg的事务是基于微批处理粒度的，相比数据库的基于数据行的粒度，显得过大，并且针对update的实现也不同于数据库的基于索引记录的更新，导致在数据更新或查询阶段需要更多的处理开销。

为方便大家交流，本号特地建立一个钉钉群，视每位粉丝如客户，欢迎扫码加入：