大数据数据湖数据工程【免费下载链接】hudiUpserts, Deletes And Incremental Processing on Big Data.项目地址https://gitcode.com/gh_mirrors/hu/hudi点击查看免费下载本篇技术指南围绕 Hudi 社区 RFC-73JIRA HUDI-6709展开系统讲解 Hudi 在单表事务能力基础上如何通过**数据库Database与数据库时间线Database Timeline**将多个表的写入收敛为单一原子事务覆盖设计动机、ACID 目标、MVCC 隔离分析、冲突解决策略、多表事务协议场景以及 SQL 写入与 DataFrame 写入两套落地方案的完整实现路径。读完本文你将理解 Hudi 多表事务的设计骨架与关键术语并能结合仓库源码定位TransactionManager、LockManager、ConflictResolutionStrategy与HoodieCatalog等核心组件为后续参与实现或二次开发打好基础。一、背景从单表事务到跨表原子性Hudi 自诞生起便强调事务保证为单个表的数据完整性与一致性提供支撑。其事务模型的核心是时间线Timeline时间线以文件系统的hoodie/目录.hoodie为载体以instant瞬时动作记录表上的每一次变更如 commit、deltacommit、compaction、clean、rollback、savepoint 等每个 instant 都带有全局单调递增的 instant time时间戳状态机在REQUESTED → INFLIGHT → COMPLETED之间流转失败/回滚则进入ROLLED_BACK或FAILED状态以时间线为唯一事实来源Hudi 通过可调的并发控制机制默认 MVCC 快照隔离 可选的外部锁让多写者、多读者在同一表上并发执行而互不干扰。这一能力的关键实现在仓库中均可找到对应代码ConflictResolutionStrategy.java冲突解决策略接口用户可实现自己的策略接口包含getCandidateInstants找出候选冲突 instant、hasConflict判断两个操作是否冲突、resolveConflict如何解决冲突、isPreCommitRequired表服务如 clustering 是否走 preCommit四个方法TransactionManager.java允许客户端开启/结束事务内部组合LockManager凡是beginTransaction与endTransaction之间的操作都保证原子性LockManager.java包装各种LockProvider实现负责锁的获取、重试与释放默认使用LOCK_ACQUIRE_CLIENT_NUM_RETRIES与LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS控制重试次数与等待时长LockProvider.java锁提供方的统一接口底层可实现为 ZooKeeper、HDFS 文件锁、本地 JVM 锁等不同形式HoodieInstant.javainstant 及其State枚举REQUESTED/INFLIGHT/COMPLETED的定义HoodieActiveTimeline.java活跃时间线实现其中的createNewInstantTime负责生成新的事务/instant 时间戳。但上述能力仅限于单表。现代数据湖架构往往由大量相互关联的表组成数据回填backfill、跨表 upsert、删除、复杂变换等操作经常横跨多张表。若某一表成功、另一表失败就会出现部分写入的脏状态。RFC-73 提出的多表事务Multi-Table Transactions正是要解决要么全部表一起提交要么全部回滚的跨表原子性问题。当前实现的四个缺口RFC-73 明确指出现有实现无法原样扩展为跨表原子操作缺少数据库概念需要先有数据库及其下属表的逻辑抽象才能把一个事务与多张表关联起来时间线需要演进Hudi 时间线必须扩展以追踪跨多张表的变更顺序这引入了跨表操作次序追踪与管理的复杂度故障面扩大涉及多表后潜在故障点成倍增加一张表的失败可能级联影响整个事务跨表回滚、尤其是存在表间依赖时的回滚一致性是新的难题并发控制不适用现有并发控制在设计上并未考虑多表事务场景。二、设计目标与非目标目标Goals目标说明跨表原子性跨多张表的一组变更要么全部完成、要么全部回滚不允许出现部分提交一致性每次事务提交后所有涉及的表都处于一致状态避免脏读、不可重复读正确处理读写与写写冲突隔离性多个并发事务互不干扰一个事务看不到另一个进行中事务的中间状态持久性多表事务一旦提交即使系统崩溃变更也永久生效且可恢复性能对常规 Hudi 操作引入的额外开销要尽量小推论OCC 场景下锁粒度不能粗于文件级可观测性与 hudi-cli 集成以跟踪事务状态与健康度提供独立的事务指标提供全面的调试日志与现有 Hudi 功能集成无缝集成现有特性不破坏既有 API 或工作流尤其保证TransactionManager、LockManager、ConflictResolutionStrategy与单表场景下的行为一致可配置性允许用户配置多表事务行为如并发控制机制、超时时间等恢复能力支持回滚、savepoint 与 restore非目标Non-goals非目标说明跨数据库事务初期不做跨数据库事务避免过度复杂记录级细粒度锁初期采用粗粒度锁如文件级不做单条记录级别的锁同时避免表级或分区级锁分布式事务不做 2PC 或 quorum 协议意义上的分布式事务假设底层存储持久且能容忍分布式故障复杂冲突解决初期实现简单策略如冲突即中止事务更复杂的策略留待后续迭代复制/CDC 流不基于数据库级提交日志构建复制或 CDC 流尽管数据库时间线在未来很适合做这件事三、核心设计数据库时间线Database Timeline解决跨表 ACID 的核心挑战RFC-73 归纳为四个关键诉求目录Catalog需要一个 catalog API 来追踪数据库及其表事务协调器Transaction Coordinator集中式协调器负责管理事务日志、跟踪进行中的事务、处理超时并管理回滚事务日志Transaction Log表级时间线已经承担了动作状态 起始/修改时间的事务日志角色数据库级则需要追踪所有多表事务——每次事务的 start、update 或 commit 都要被记录。以数据库时间线 各表时间线配合可以保证数据文件在事务的全部语句完成之前不可见锁/冲突解决机制在 OCC 外部锁提供方的模式下多表事务期间锁定受影响的文件防止冲突写入并决定冲突解决策略如 last write wins、version vectors。注意Hudi 默认基于 MVCC 为读写之间提供快照隔离读者无需加锁。数据库时间线包含的信息数据库级时间线作为事务日志主要记录Transaction ID事务启动时刻的 instant timeEnd/Modified Timestamp结束/修改时间戳事务状态REQUESTED、INFLIGHT、COMPLETED、ROLLED_BACK其他相关元数据这与单表时间线中 instant 的状态机REQUESTED → INFLIGHT → COMPLETED见 HoodieInstant.java同构——事务被建模为数据库级的instant从而复用 Hudi 成熟的时间线语义与持久化方式。事务 ID 直接使用HoodieActiveTimeline.createNewInstantTime()生成的单调递增时间戳保证跨事务的全局有序。四、MVCC 快照隔离与异常分析Hudi 天然支持基于 MVCC 的快照隔离配合表时间线与数据库时间线即可在并发读写下为多表事务提供快照隔离。其读写模型是事务开始事务启动时被赋予一份数据库启动时刻的快照读取事务从自己的快照读取看不到其他事务的中间变更写入写入创建数据的新版本不影响其他进行中的事务它们继续使用各自的快照提交提交时检查写写冲突——若另一事务在当前事务启动时间之后已提交了对同一数据的修改则当前事务可能被中止。四类并发异常逐一分析1. 脏读Dirty Reads——事务读到了仍在进行中事务写入的数据。 MVCC 下天然被阻止每个事务都从自己的快照读取看不到其他事务未提交的变更。2. 幻读Phantom Reads——事务执行过程中新记录被添加或旧记录被删除恰好符合该事务此前某次读取的过滤条件。 由于事务基于快照操作看不到快照之后新增的记录。若事务意图是保证某类新记录不被加入则需要额外机制如谓词锁。如果事务内部不允许刷新快照则幻读不可能发生包括自连接场景。3. 读写冲突Read-Write Conflict——事务读取多项数据时另一事务修改了其中先读的部分导致不一致视图。 MVCC 下不发生异常事务读取的数据在事务持续期间保持一致读取看到的是与读操作开始时间一致的快照不会看到未提交变更。4. 写写冲突Write-Write Conflict——两个事务基于各自读取决定修改同一数据最终结果可能与任一事务的意图都不一致。 这是即使在快照隔离下也可能存在的问题。例如 T1、T2 同时启动并读取同一表中的同一文件二者都基于各自快照决定修改该文件互不可见对方变更提交时都试图创建该文件的新版本于是产生写写冲突。系统检测到冲突后通常会中止其中一个事务或应用某种冲突解决策略。五、冲突检测与三种解决策略冲突检测基于事务变更的分区 文件IDfileId集合进行。设想两个冲突事务 T1、T2它们启动时都拉取所有表的最新版本并以其启动时间戳在数据库时间线上注册该时间戳大于 T1、T2 涉及的所有表版本。提交前的冲突处理有三条可选路线策略一先提交者 / 年轻事务胜出First committer / Younger transaction wins本质上无需额外解析仅按结束时间戳排序。若某事务试图修改已被更年轻事务时间戳更晚修改过的数据则更老的事务被回滚以避免冲突从而按时间戳对事务完成串行化。优点事务按一致顺序串行化在低竞争环境中表现良好。缺点必须为每个动作维护结束时间戳高竞争环境下中止率可能很高。策略二OCC带锁提供方 wait-die / wound-wait借助起始与结束时间戳可实现经典数据库中的 wait-die 与 wound-wait 策略参考 CMU 15-445 关于两阶段锁的讲义Wait-Die若较老的事务请求被较年轻事务持有的锁较老事务被迫等待超时后死亡被中止Wound-Wait较年轻事务被打伤中止让较老事务继续执行。优点提供了优先较老事务的机制。缺点实现复杂若频繁出现短事务等待时间可能很高。策略三补偿事务Compensating Transaction检测到冲突时不中止原事务而是基于最新快照执行另一事务以补偿冲突事务产生的影响参考 Azure 架构中心的 compensating-transaction 模式。优点避免中止与重试适合长事务。缺点设计与保证补偿事务正确性的复杂度高。从现状看仓库中已有的默认实现 SimpleConcurrentFileWritesConflictResolutionStrategy.java 正是简单冲突解决的雏形它通过getCandidateInstants筛选出上次成功写入之后完成的 commit 以及当前 instant 之后调度/完成的 compaction、clustering 动作作为候选再通过比较两个操作getMutatedPartitionAndFileIds()返回的分区文件ID集合是否有交集来判断冲突一旦交集非空resolveConflict直接抛出HoodieWriteConflictExceptionCannot resolve conflicts for overlapping writes即冲突即中止——这与 RFC-73 非目标中初期采用简单策略冲突即中止的取向一致。可据此推断多表事务初期的冲突解决大概率沿用这一集合交集判冲突、冲突即中止的模式。六、多表事务协议四个典型场景以下协议场景均假设存在逻辑上的数据库概念每个数据库拥有自己的数据库时间线用于追踪该数据库内跨表的事务。场景一涉及三张表更新的基本事务开启事务写者W1发起事务数据库时间线以唯一事务 IDTxID1记录新事务的开始。更新各表W1向t1写入更新更新被暂存staged并在t1时间线中以TxID1关联、状态为 pending同样向t2、t3写入各自表时间线中关联TxID1。提交事务W1发出事务结束信号数据库时间线将事务TxID1标记为已提交t1、t2、t3的时间线将TxID1下的 pending 更新落定finalize。场景二单表出现冲突的事务开启事务W1发起事务TxID1另一写者W2发起事务TxID2。冲突更新W1与W2同时尝试写入t1中的同一共享文件当W2尝试提交时MVCC 系统检测到冲突因为W1已暂存了更新W2对t1的变更被中止W2收到冲突通知。场景三pending 更新期间的并发读开启事务W1发起事务TxID1暂存t1、t2的更新t3的更新仍在进行。并发读取读者R1想读t1、t2、t3。由于W1对t1、t2的更新只是暂存、事务尚未提交R1在这些表上看到的是数据旧状态对t3由于W1尚未暂存或提交任何更新R1读到当前状态。场景四更新失败需要回滚的事务开启事务W1发起事务TxID1成功暂存t1、t2的更新。更新失败对t3的更新因内部错误如 I/O 错误失败。回滚事务W1对TxID1发起回滚数据库时间线将事务TxID1标记为 rolled backt1、t2时间线中TxID1关联的暂存更新被丢弃。协议要点上述流程保证了快照隔离——读者看到的永远是一致的整体状态事务更新之前或之后绝不会看到二者的混合态。七、实现方案一SQL 写入Spark Catalog 扩展RFC-73 给出的 SQL 语义草案如下BEGIN tx1 // anything done here is associated with tx1. // load table A; (load As latest snapshot time into HoodieCatalog/driver memory) // load table B; (load Bs latest snapshot time into HoodieCatalog/driver memory) // load table A again; (it will reuse the snapshot time already in HoodieCatalog/driver memory). COMMIT / ROLLBACK其中有两个待解决的关键问题能否在 Spark 的HoodieCatalog内实现这一能力如何识别进入 Catalog 的请求属于哪个事务实现路线分为四步语法层为 BEGIN...END 块编写语法规则并同步修改 parser 与 lexer——Hudi 此前实现MERGE INTO时已有类似经验。抽象层新建TransactionalHoodieCatalog实现 Spark 的TableCatalog接口并叠加多表事务能力状态管理在 namespace 内维护当前事务及其涉及的表集合持久化到文件系统的数据库时间线事务方法新增beginTransaction、commitTransaction等方法让createTable、loadTable具备事务感知其他需要的方法同理Catalog 方法实现TableCatalog接口中关于 namespace 与表的其他方法。命令层新增BeginMultiTableTransactionCommand当 parser 判定语句为 BEGIN...END 块时调用 catalog 的beginTransaction发起事务块内每个 DML 语句照常经HoodieSparkSqlWriter执行在执行任何变更前调用catalog.addTableToTransaction当 parser 碰到 END 结束事务作用域时由另一个命令EndMultiTableTransactionCommand调用catalog.commitTransaction()落盘全部变更或调用rollbackTransaction()丢弃它们。读写 API 兼容Spark 的TableCatalog是 DataSource V2 API因此写入仅支持 DSv2 路径读取则回退到 V1——这与今天HoodieCatalog的做法一致。RFC-73 中给出的TransactionalHoodieCatalog骨架代码如下要点已标注abstract class TransactionalHoodieCatalog extends TableCatalog { val spark: SparkSession SparkSession.active // Representing a Transaction case class Transaction(transactionId: String, tablesInvolved: mutable.Set[String]) // State private var currentTransaction: Option[Transaction] None // Databases and Tables - in-memory representation for this sketch private val databases: mutable.Map[String, CatalogDatabase] mutable.Map() private val tables: mutable.Map[String, CatalogTable] mutable.Map() // Transactional methods def beginTransaction(): String { val newTransactionId HoodieActiveTimeline.createNewInstantTime() currentTransaction Some(Transaction(newTransactionId, mutable.Set())) newTransactionId } def addTableToTransaction(tableName: String): Unit { currentTransaction match { case Some(tx) tx.tablesInvolved tableName case None throw new IllegalStateException(No active transaction to add table to.) } } def commitTransaction(): Unit { // Commit logic currentTransaction None } def rollbackTransaction(): Unit { // Rollback logic currentTransaction None } // Implementation of Catalog methods // Database related operations override def createDatabase(dbDefinition: CatalogDatabase, ignoreIfExists: Boolean): Unit ??? override def dropDatabase(db: String, ignoreIfNotExists: Boolean, cascade: Boolean): Unit ??? override def alterDatabase(dbDefinition: CatalogDatabase): Unit ??? override def getDatabase(db: String): CatalogDatabase { databases.getOrElse(db, throw new NoSuchDatabaseException(db)) } override def databaseExists(db: String): Boolean { databases.contains(db) } override def listDatabases(): Seq[String] { databases.keys.toSeq } override def listDatabases(pattern: String): Seq[String] { databases.keys.filter(_.matches(pattern)).toSeq } // ... More database methods // Table related operations override def createTable(tableDefinition: CatalogTable, ignoreIfExists: Boolean): Unit { // If theres an ongoing transaction, associate the table with it currentTransaction.foreach(tx tx.tablesInvolved tableDefinition.identifier.table) // Rest of the creation logic tables (tableDefinition.identifier.table - tableDefinition) } override def dropTable(db: String, table: String, ignoreIfNotExists: Boolean, purge: Boolean): Unit ??? override def loadTable(db: String, table: String, loadPath: String, isOverwrite: Boolean, isSrcLocal: Boolean): Unit { tables.getOrElse(table, throw new NoSuchTableException(db, table)) } }其中beginTransaction使用HoodieActiveTimeline.createNewInstantTime()生成事务 ID——该静态方法正是 HoodieActiveTimeline.java 中基于HoodieInstantTimeGenerator生成全局单调时间戳的入口。关于 DSv1 / DSv2 回退的取舍Hudi 的 relation 目前不实现 DSv2 读 API只支持 DSv2 写 API。LHBench 论文CIDR 2023指出DSv2 API 尚不成熟且不报告查询规划所需的部分指标可能导致比 Iceberg 更差的查询计划。而写入侧 Hudi 本身也会回退到 DSv1schema-on-read 场景除外。这一点在仓库中有直接代码证据在 HoodieCatalog.scala 的loadTable中默认将 Hudi 表包装为v1TableWrapper回退到 V1 读取仅在开启 schema evolution 时才返回v2Table对应 HUDI-4178。为什么需要回退虽然 DataFrame shuffle 主要由 Spark 的逻辑/物理规划阶段决定DSv2 提供了SupportsRead、SupportsWrite及其子接口以支持更好的下推与优化合理利用可以减少 shuffle但 DSv2 并不直接提供自定义 shuffle 逻辑的接口。数据源只能通过SupportsReportPartitioningtrait 上报自身物理分布如 bucketing、partitioningSpark 据此避免不必要的 shuffle若数据源未正确实现或上报这些特征可能导致次优执行计划——因此责任部分在于数据源实现的准确性。多表事务方案沿用同样思路在事务 catalog 中创建 v2Table写入时回退到V1Write。八、实现方案二DataFrame 写入HoodieTransactionManager对于编程式 APIRFC-73 提出的方案是通过HoodieTransactionManager将多个 DataFrame 写入纳入同一事务// similar to catalog.beginTransaction val transaction HoodieTransactionManager.beginTransaction() // similar to catalog.addTableToTransaction and execute the write. // Then pass the control to HoodieTransactionManager. df1.write.format(hudi).withTransaction(transaction).options(opts1).save(basePath1) // Same as above. If this failed, then rollback and pass the control to HoodieTransactionManager. // HoodieTransactionManager will rollback all previous transactions associated with transaction id. df2.write.format(hudi).withTransaction(transaction).options(opts2).save(basePath2) // All success, go ahead and commit or finish rollback (ROLLED_BACK state on timeline is written in that case) transaction.commit()流程要点beginTransaction()对应 catalog 方案中的beginTransaction返回事务句柄每个withTransaction(transaction)的写操作对应addTableToTransaction 实际写入任一写入失败控制权交回HoodieTransactionManager它会按事务 ID 回滚此前所有关联写入全部成功则transaction.commit()提交若走回滚路径则在时间线上写入ROLLED_BACK状态。值得注意的是回滚事务、跟踪当前事务拥有者这一职责在现有代码中已有对应基础单表场景的 TransactionManager.java 通过currentTxnOwnerInstant/lastCompletedTxnOwnerInstant记录事务拥有者与最近一次成功事务并在beginTransaction时按需加锁、endTransaction时释放。RFC-73 目标中明确要求多表实现不破坏这些 API 的单表语义因此HoodieTransactionManager可理解为将此类能力从单表 instant 级提升到跨表事务级的扩展。九、落地与采用计划对现有用户的影响设计建立在时间线格式变更的前提上。用户可以在升级版本后新建的表上使用新特性旧数据集未包含 database / database timeline 概念仍应保持可读。希望为既有数据集启用多表事务的用户则需要平滑的迁移路径——可能借助工具或脚本将旧时间线事件重构/迁移到新的数据库时间线中具体迁移工具在 RFC 中标记为 TODO。旧行为如何退出无需淘汰旧行为新行为通过显式使用新 API 或 SQL 语法激活老路径完全不受影响。十、测试计划RFC-73 规划了四类测试新 API 与 SQL 语法的单元测试新时间线格式的单元测试冲突解决的单元测试含失败场景与正常场景的多表事务集成测试。从仓库现状看并发写入与冲突解决的既有测试矩阵如 hudi-client 与 hudi-spark-datasource 下的 functional 测试可作为上述测试的继承基础多表事务的测试将在此之上补充跨表维度。附录教科书式幻读示例设有一张Employees表含employee_id与manager_id两列其中部分员工的manager_id指向表中其他员工的employee_id。自连接可查询员工与其管理者的对应关系SELECT e1.employee_id, e2.employee_id AS manager_id FROM Employees e1 JOIN Employees e2 ON e1.manager_id e2.employee_id;事务 A启动并执行上述自连接得到员工与管理者列表事务 A 进行期间事务 B启动并向Employees表插入新行一个新员工其管理者的employee_id已存在于表中事务 B提交若事务 A重跑同一自连接可能看到新插入的行导致 join 输出多出一条初次查询时没有的结果——这就是幻读。在 MVCC 快照隔离级别下事务即使重跑自连接仍只看到事务启动时的数据库状态因此可阻止该幻读但读已提交read committed级别无法保证这一点。这也正是 RFC-73 坚持事务内不刷新快照以彻底杜绝幻读的原因。参考资料RFC 原文rfc-73.md冲突解决策略接口ConflictResolutionStrategy.java默认冲突解决实现SimpleConcurrentFileWritesConflictResolutionStrategy.java事务管理器TransactionManager.java锁管理器与锁接口LockManager.java、LockProvider.java时间线相关HoodieActiveTimeline.java、HoodieInstant.javaSpark Catalog 现状HoodieCatalog.scala、HoodieInternalV2Table.scala赞分享大数据数据湖数据工程【免费下载链接】hudiUpserts, Deletes And Incremental Processing on Big Data.项目地址https://gitcode.com/gh_mirrors/hu/hudi点击查看免费下载相关推荐Apache Hudi 多表事务RFC-73深度解析数据库时间线驱动的跨表 ACID 设计与实现方案Apache Hudi 多表事务RFC 73深度解析数据库时间线驱动的跨表 ACID 设计与实现方案 本文以 Apache Hudi RFC 73Mul数据湖湖仓一体大数据数据存储Jedis 事务Transactions/MULTI实战指南从立即启动到 WATCH 乐观锁与多数据库事务Jedis 事务Transactions/MULTI实战指南从立即启动到 WATCH 乐观锁与多数据库事务 导读 本文以 Jedis当前仓库为 gh_m数据库缓存后端Apache Pulsar 事务Transactions核心概念详解事务协调器、TxnID、事务缓冲与 Pending Ack 状态Apache Pulsar 事务Transactions核心概念详解事务协调器、TxnID、事务缓冲与 Pending Ack 状态 导读 本文基于 A消息队列后端流处理上一篇mlcourse.ai 的 Python 前置技能要求与自测指南从基础语法到从零实现 SGD下一篇Equalizer APO Stage预混/后混深度解析upmix与设备级过滤的正确姿势创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考