一个数据处理任务在凌晨三点失败了。它成功写入了1TB数据中的700GB到数据湖，然后因为一个暂时的网络故障而崩溃。现在，数据湖处于一个不一致的、损坏的状态。下游的报表和机器学习模型读取了这些不完整的数据，造成了整个业务线的混乱。这种场景是原始数据湖架构的常态，缺乏事务保证是其核心痛点。我们需要的是在对象存储这种天生非事务性的媒介之上，构建一个能够提供原子性和一致性的元数据层。
定义架构挑战：在对象存储上实现ACID
对象存储（如AWS S3）是构建数据湖的事实标准，它提供了近乎无限的扩展性、高持久性和低成本。但它的API极其简单：put, get, delete。它不是数据库，没有事务、没有原子性的多文件操作。我们的核心技术问题是：如何在一个由不可变文件组成的数据湖中，实现类似数据库的原子提交（Atomicity）和快照隔离（Isolation）？
这本质上要求我们设计一个独立的、可靠的元数据服务来追踪构成一个逻辑数据表的所有物理文件。当一个写操作（例如，一个Spark作业）完成时，它需要以原子方式更新元数据，要么所有新文件的引用被一次性添加，要么什么都不发生。这直接将我们引向了分布式系统设计的核心——CAP理论的权衡。
方案A：基于强一致性协调器的CP架构
最直接的思路是引入一个外部的、支持事务的强一致性系统来管理元数据。
架构设计:
- 使用一个集中的关系型数据库（如PostgreSQL）或一个分布式共识系统（如etcd、ZooKeeper）作为元数据存储。
- 元数据存储中包含两类核心信息：
- Table Versions: 一张表记录了每个逻辑表的当前版本号。
- File Manifest: 另一张表记录了每个版本号具体包含哪些数据文件（S3上的路径）。
- 写操作流程:
- 一个写任务在S3的临时目录中生成新的数据文件。
- 任务向元数据存储发起一个数据库事务。
- 在事务内，锁定该表的版本记录。
- 将新生成的数据文件路径插入File Manifest表，关联到一个新的版本号。
- 更新Table Versions表中该表的版本号。
- 提交数据库事务。
优劣分析:
- 优点: 实现逻辑相对简单。利用了外部系统的ACID保证，我们可以轻松获得强一致性（Consistency）。任何时候，客户端查询元数据总能得到一个绝对正确的、一致的表状态。
- 缺点: 这引入了一个明显的瓶颈和单点故障。元数据存储的可用性（Availability）和性能直接决定了整个数据湖的写入吞吐量。在网络分区导致写任务无法连接到元数据存储时，整个写入流程将完全停滞。这是一个典型的CP（Consistency-Partition Tolerance）系统，为了保证一致性，牺牲了部分场景下的可用性。对于需要高并发写入的数据平台，这种架构很快会达到瓶颈。
sequenceDiagram
participant Writer as Writer Job
participant S3
participant RDBMS as PostgreSQL (CP Coordinator)
Writer->>S3: 1. Write new data files to temp location
activate S3
S3-->>Writer: Files created
deactivate S3
Writer->>RDBMS: 2. BEGIN TRANSACTION
activate RDBMS
RDBMS-->>Writer: OK
Writer->>RDBMS: 3. LOCK table_versions FOR UPDATE
RDBMS-->>Writer: Lock acquired
Writer->>RDBMS: 4. INSERT INTO file_manifest (path, version)
RDBMS-->>Writer: Rows inserted
Writer->>RDBMS: 5. UPDATE table_versions SET version = version + 1
RDBMS-->>Writer: Row updated
Writer->>RDBMS: 6. COMMIT
RDBMS-->>Writer: Transaction committed
deactivate RDBMS
方案B：基于对象存储原子操作的乐观并发AP架构
另一个截然不同的思路是放弃中心化协调器，直接利用对象存储本身的一些特性来构建一个去中心化的、高可用的事务协议。
架构设计:
- 日志式元数据: 我们将表的元数据历史记录本身也存储在S3上，形成一个不可变的提交日志（Commit Log）。例如，在表路径下创建一个
_commit_log目录。 - 原子提交: 每次提交，我们都会尝试原子性地创建一个新的JSON或Avro文件，文件名代表版本号，例如
000000.json,000001.json… 这个文件的内容描述了本次事务所做的变更（比如，添加了哪些文件，删除了哪些文件）。 - 利用原子性: 关键在于如何“原子性地创建”这个文件。大多数对象存储都提供某种形式的“put-if-absent”或条件写入（conditional put）功能。这意味着我们可以发起一个请求：“只有当目标路径
_commit_log/000001.json不存在时，才创建它”。这个操作在对象存储层面是原子的。 - 写操作流程 (Optimistic Concurrency Control):
- 一个写任务首先需要确定当前表的最新版本号N。它通过列出
_commit_log目录下的所有文件来实现。 - 任务在S3的临时目录生成新的数据文件。
- 任务构建一个描述本次变更的JSON对象，并尝试使用“put-if-absent”操作将其写入
_commit_log/N+1.json。 - 冲突检测:
- 如果写入成功，则事务提交成功。
- 如果写入失败（因为文件已存在），这说明在当前任务执行期间，有另一个任务已经成功提交了版本N+1。这就是一个写冲突。
- 重试: 发生冲突后，任务必须中止当前操作，回退到第一步，重新读取最新的版本号（现在是N+1），然后重试整个流程。
- 一个写任务首先需要确定当前表的最新版本号N。它通过列出
- 日志式元数据: 我们将表的元数据历史记录本身也存储在S3上，形成一个不可变的提交日志（Commit Log）。例如，在表路径下创建一个
优劣分析:
- 优点: 极高的可扩展性和可用性（Availability）。没有中心化的瓶颈，写入吞吐量可以随着计算集群的规模线性扩展。即使在部分网络分区下，只要任务能连接到S3，就可以尝试提交。这更像一个AP（Availability-Partition Tolerance）系统，它保证了系统的可用性，并通过乐观锁和重试机制最终达到一致性。
- 缺点: 实现逻辑更复杂。需要处理并发冲突和重试循环。读取方（下游消费者）的逻辑也更复杂，它们需要知道如何通过读取提交日志来重构出表的最新状态。在写入冲突非常频繁的场景下，性能可能会因为不断的重试而下降。
sequenceDiagram
participant Writer1 as Writer Job 1
participant Writer2 as Writer Job 2
participant S3
Writer1->>S3: 1. List _commit_log, find latest is N
Writer2->>S3: 1. List _commit_log, find latest is N
Writer1->>S3: 2. Write new data files
Writer2->>S3: 2. Write new data files
Writer1->>S3: 3. ATTEMPT: Create _commit_log/N+1.json (put-if-absent)
activate S3
S3-->>Writer1: SUCCESS
deactivate S3
Note over Writer1: Commit successful!
Writer2->>S3: 4. ATTEMPT: Create _commit_log/N+1.json (put-if-absent)
activate S3
S3-->>Writer2: FAILED (file already exists)
deactivate S3
Note over Writer2: Conflict detected!
Writer2->>S3: 5. RETRY: List _commit_log, find latest is N+1
Writer2->>S3: 6. ATTEMPT: Create _commit_log/N+2.json (put-if-absent)
activate S3
S3-->>Writer2: SUCCESS
deactivate S3
Note over Writer2: Retry successful!
最终选择与理由：拥抱去中心化的AP架构
在真实的数据湖仓场景中，通常有大量的ETL、流式摄入等任务并发写入。对写入路径的可用性和可扩展性的要求，远高于对元数据读取的强一致性要求。一个短暂的元数据不一致（例如，一个分析任务读取到几秒钟前的表状态）通常是可以接受的，但一个因为中心化锁竞争而阻塞所有写入任务的系统是不可接受的。
因此，我们选择方案B。它将CAP的权衡倾向了A（可用性）和P（分区容忍性），同时通过客户端的乐观并发协议来保证最终的数据一致性和事务原子性。Clojure的不可变数据结构和函数式编程范式，非常适合处理这种基于不可变日志的、无副作用的事务逻辑。
核心实现：Clojure中的乐观并发事务层
我们将使用Clojure构建这个事务层的核心逻辑。假设我们使用AWS S3作为底层存储。
项目依赖与配置
首先，定义项目依赖。我们需要AWS Java SDK v2、一个JSON库和日志库。
deps.edn:
{:paths ["src"]
:deps {org.clojure/clojure {:mvn/version "1.11.1"}
org.clojure/data.json {:mvn/version "2.4.0"}
org.clojure/tools.logging {:mvn/version "1.2.4"}
ch.qos.logback/logback-classic {:mvn/version "1.3.11"}
;; AWS SDK v2 for S3
software.amazon.awssdk/s3 {:mvn/version "2.20.162"}}}底层S3原子操作封装
我们需要一个函数来执行put-if-absent。S3本身没有直接的put-if-absent API，但我们可以通过If-None-Match: *请求头来模拟，它表示“如果目标对象不存在（ETag不匹配任何值），则执行操作”。
(ns lakehouse.s3
(:require [clojure.tools.logging :as log])
(:import (software.amazon.awssdk.core.sync RequestBody)
(software.amazon.awssdk.services.s3 S3Client)
(software.amazon.awssdk.services.s3.model PutObjectRequest ConditionalCheckFailedException NoSuchKeyException ListObjectsV2Request)
(java.nio.charset StandardCharsets)))
(defonce ^:private s3-client (-> (S3Client/builder)
(.build)))
(defn put-object-if-absent!
"Atomically puts an object to S3 if it does not already exist.
Returns true on success, false on conflict (object already exists)."
[^String bucket ^String key ^String content]
(let [request (-> (PutObjectRequest/builder)
(.bucket bucket)
(.key key)
; If-None-Match: * ensures atomicity.
; The request will fail with a 412 PreconditionFailed
; if the object already exists.
(.ifNoneMatch "*")
(.build))
body (RequestBody/fromString content StandardCharsets/UTF_8)]
(try
(.putObject s3-client request body)
(log/info "Successfully wrote" key "to bucket" bucket)
true
(catch ConditionalCheckFailedException e
; This is the expected exception for a commit conflict. It's not an error.
(log/warn "Commit conflict: object" key "already exists in bucket" bucket)
false)
(catch Exception e
(log/error e "Failed to write object to S3")
(throw e)))))
(defn list-object-keys
"Lists all object keys under a given prefix."
[^String bucket ^String prefix]
(try
(let [request (-> (ListObjectsV2Request/builder)
(.bucket bucket)
(.prefix prefix)
(.build))
response (.listObjectsV2 s3-client request)]
(map #(.key %) (.contents response)))
(catch NoSuchKeyException _
;; This is fine, means the prefix doesn't exist yet
[])))这里的错误处理很关键。ConditionalCheckFailedException (HTTP 412) 不是一个需要上报的系统错误，而是我们并发控制协议中的一个预期信号，代表了写冲突。
事务日志与提交逻辑
现在，我们来构建核心的事务提交逻辑。
(ns lakehouse.transaction
(:require [lakehouse.s3 :as s3]
[clojure.data.json :as json]
[clojure.string :as str]
[clojure.tools.logging :as log]))
(def ^:private commit-log-prefix "_commit_log/")
(defn- get-latest-version
"Finds the latest version number from the commit log."
[^String bucket ^String table-path]
(let [full-prefix (str table-path commit-log-prefix)
log-files (s3/list-object-keys bucket full-prefix)]
(if (empty? log-files)
-1 ; Sentinel for a new table
(->> log-files
(map #(str/replace % full-prefix ""))
(map #(str/replace % #".json" ""))
(map #(try (Long/parseLong %) (catch Exception _ -1)))
(apply max)))))
(defn- format-version [version]
(format "%020d.json" version))
(defrecord Transaction [bucket table-path actions])
(defn create-transaction
"Starts a new transaction for a given table."
[bucket table-path]
(->Transaction bucket table-path []))
(defn add-file
"Stages a new file addition to the transaction."
[transaction file-path]
(update transaction :actions conj {:add {:path file-path}}))
(defn commit!
"Attempts to commit the transaction using optimistic concurrency.
Retries on conflict up to a certain limit."
[transaction {:keys [max-retries retry-interval-ms]
:or {max-retries 10 retry-interval-ms 1000}}]
(loop [attempt 1]
(if (> attempt max-retries)
(throw (ex-info "Failed to commit transaction after max retries"
{:transaction transaction, :max-retries max-retries}))
(let [current-version (get-latest-version (:bucket transaction) (:table-path transaction))
next-version (inc current-version)
commit-file-key (str (:table-path transaction) commit-log-prefix (format-version next-version))
commit-content (json/write-str {:version next-version, :actions (:actions transaction)})]
(log/infof "Attempt %d/%d: Committing version %d for table %s"
attempt max-retries next-version (:table-path transaction))
(if (s3/put-object-if-absent! (:bucket transaction) commit-file-key commit-content)
{:status :success, :version next-version}
;; Conflict detected, wait and retry
(do
(log/warnf "Attempt %d/%d: Conflict detected. Retrying in %d ms..."
attempt retry-interval-ms)
(Thread/sleep retry-interval-ms)
(recur (inc attempt))))))))commit! 函数是整个架构的核心。它封装了“读取-修改-写入”的循环，并在写入失败时自动重试。这种声明式的、带有重试逻辑的实现，在Clojure中用 loop/recur 表达得非常自然。
示例：并发写入
下面是一个模拟两个并发作业写入的例子。
(defn run-writer-job
[job-id bucket table-path]
(let [file-to-add (format "data-%s-%s.parquet" job-id (System/currentTimeMillis))
tx (-> (create-transaction bucket table-path)
(add-file file-to-add))]
(log/infof "[Job %s] Starting commit..." job-id)
(let [result (commit! tx {:max-retries 5 :retry-interval-ms 500})]
(log/infof "[Job %s] Commit successful! New version: %d" job-id (:version result)))))
;; --- Main execution ---
(let [bucket-name "my-lakehouse-bucket"
table-name "sales_data/"]
;; For demonstration, you would need to have an S3 bucket setup.
;; The following code is illustrative.
(log/info "--- Starting concurrent write simulation ---")
(let [writer1 (future (run-writer-job "A" bucket-name table-name))
writer2 (future (run-writer-job "B" bucket-name table-name))]
@writer1
@writer2)
(log/info "--- Simulation finished ---"))
;; Expected Log Output:
;; INFO [Job A] Starting commit...
;; INFO [Job B] Starting commit...
;; INFO [main] Attempt 1/5: Committing version 0 for table sales_data/
;; INFO [main] Attempt 1/5: Committing version 0 for table sales_data/
;; INFO [main] Successfully wrote sales_data/_commit_log/00000000000000000000.json to bucket my-lakehouse-bucket
;; INFO [Job A] Commit successful! New version: 0
;; WARN [main] Commit conflict: object sales_data/_commit_log/00000000000000000000.json already exists in bucket my-lakehouse-bucket
;; WARN [main] Attempt 1/5: Conflict detected. Retrying in 500 ms...
;; INFO [main] Attempt 2/5: Committing version 1 for table sales_data/
;; INFO [main] Successfully wrote sales_data/_commit_log/00000000000000000001.json to bucket my-lakehouse-bucket
;; INFO [Job B] Commit successful! New version: 1这个模拟清晰地展示了乐观锁机制：一个作业成功，另一个检测到冲突、读取新版本并成功重试。我们成功地在S3上实现了原子提交，而无需任何外部协调服务。
架构的扩展性与局限性
我们构建的这套基于乐观并发的事务层，为数据湖提供了强大的原子性保证，但它并非银弹。在真实项目中，这个基础之上还需要考虑更多。
一个明显的局限性是性能。每次读取最新版本都需要列出所有提交日志文件，当日志文件成千上万时，这个操作会变得很慢。生产级系统（如Delta Lake, Apache Iceberg）通过Checkpointing机制来解决这个问题：定期将一长串JSON日志文件合并成一个高效的Parquet文件（快照），这样客户端只需读取最新的Checkpoint文件和之后少量的JSON增量日志即可。
另一个问题是高频写入下的冲突率。如果几十个任务同时写入一张小表，可能会导致大部分任务都在不断重试，浪费计算资源。这表明该架构更适合并发度适中、写入延迟不那么敏感的批处理场景，而非高频交易系统。
最后，我们的实现只解决了写入方的原子性。读取方如何消费这些数据（即如何发现并可靠地只读取某个一致性版本的数据）是另一个复杂的话题，需要客户端库的支持，确保它们能正确解析提交日志并只读取有效的数据文件。但这套日志系统为实现“时间旅行”（Time Travel）查询——即查询表在过去某个版本或某个时间点的状态——打下了坚实的基础。
