構建基於事件溯源與Apache Spark的實時讀模型及其端到端測試驗證

分布式架構

文章字数: 3.9k

阅读时长: 16 分

1. 定義問題：耦合的讀寫模型與性能瓶頸

在設計一個高吞吐量的數據密集型應用時，一個核心的矛盾經常出現：寫入操作要求數據的高度規範化與事務一致性，以保證業務邏輯的準確無誤；而讀取操作，特別是複雜的分析與聚合查詢，則希望數據是反規範化的、預計算的，以便快速響應。

傳統的單體數據庫架構試圖用同一套模型滿足這兩種截然不同的需求。其典型方案是：

方案 A：單一數據庫，讀寫一體
- 實現：所有業務操作直接讀寫同一個關係型數據庫（如 PostgreSQL）。寫入時遵循嚴格的範式，通過事務保證一致性。讀取時，通過複雜的 JOIN、GROUP BY，甚至數據庫觸發器、物化視圖來生成報表或聚合數據。
- 優點：
  1. 強一致性：ACID 事務提供了最直接的數據一致性保證。
  2. 技術棧簡單：開發與運維團隊對關係型數據庫非常熟悉。
- 缺點：
  1. 性能瓶頸：高併發的寫入會與耗時的讀取查詢產生鎖競爭，導致性能急劇下降。
  2. 擴展性差：數據庫成為整個系統的中心瓶頸，垂直擴展成本高昂，水平擴展（分片）則極其複雜且具有侵入性。
  3. 模型僵化：讀取需求的任何變更都可能需要修改核心的數據表結構（ALTER TABLE），這在大型系統中是高風險操作。物化視圖的邏輯也與核心表緊密耦合。

這種架構在業務初期或許堪用，但隨着流量增長和分析維度的增加，很快就會觸及天花板。系統將變得緩慢、脆弱且難以維護。

2. 架構選型：CQRS 與事件溯源的引入

為了解決上述矛盾，我們需要將讀寫模型進行物理隔離。命令查詢職責分離（CQRS）模式是解決這個問題的指導思想。結合事件溯源（Event Sourcing），我們可以構建一個更具彈性、可擴展性和可觀測性的系統。

方案 B：基於事件溯源的 CQRS 架構
- 實現：
  1. **命令端 (Write Side)**：不再直接修改狀態。而是接收命令（Command），驗證後生成一個或多個不可變的事件（Event），並將這些事件持久化到一個僅追加的日誌中（即事件存儲）。系統的當前狀態是通過重放所有歷史事件計算得出的。
  2. **事件存儲 (Event Store)**：通常使用消息隊列（如 Kafka）或專用事件數據庫實現。它是系統唯一的事實來源（Single Source of Truth）。
  3. **查詢端 (Read Side)**：獨立的進程（投影器，Projector）訂閱事件流。每當接收到新事件時，它會更新一個或多個專為查詢優化的讀模型（Read Model）。這些讀模型可以是 Redis 中的緩存、Elasticsearch 中的索引，或數據倉庫中的聚合表。
  4. 數據流：用戶操作觸發命令 -> 命令服務生成事件 -> 事件存入 Kafka -> Apache Spark Streaming 作業消費事件 -> 更新優化的讀模型（例如存儲在 Cassandra 或 Redis 中） -> 查詢服務從讀模型中讀取數據。
- 優點：
  1. 獨立擴展：寫入吞吐量可以通過擴展命令服務和 Kafka 分區來提升。讀取性能可以通過優化讀模型和擴展查詢服務來提升。兩者互不影響。
  2. 模型靈活性：可以隨時增加新的讀模型來滿足新的查詢需求，只需從頭開始重放事件流即可生成，無需改動核心寫入邏輯。
  3. 內置審計日誌：事件流本身就是一個完美的、不可變的業務操作審計日誌。
  4. 時間旅行：可以重現系統在任意時間點的狀態，對調試和業務分析極具價值。
- 缺點：
  1. 最終一致性：從事件產生到讀模型更新存在延遲。這要求業務和產品設計能夠接受這種延遲。
  2. 架構複雜度：引入了消息隊列、流處理引擎等多個組件，增加了開發和運維的複雜度。
  3. 事件模式演進：對事件的 Schema 進行變更是一個需要謹慎處理的挑戰。

決策理由：對於需要處理海量實時數據、支持複雜多維度分析且對未來擴展性有極高要求的場景，方案 B 的長期收益遠大於其複雜性帶來的成本。它從根本上解決了讀寫衝突，為系統的演進提供了堅實的基礎。

3. 核心實現：VPC 內的數據流動

我們的目標是構建一個實時用戶行為分析儀表盤。用戶在前端的每一次關鍵操作（如點擊、購買）都將作為事件被捕獲、處理並最終反映在儀表盤上。

3.1. 網絡基礎設施：VPC 隔離與安全組

在生產環境中，所有組件必須運行在一個安全的、隔離的網絡環境中。AWS VPC 是實現這一點的基礎。

graph TD
    subgraph "VPC (10.0.0.0/16)"
        subgraph "Public Subnet (API Gateway / Load Balancer)"
            ALB[ALB/API Gateway]
        end

        subgraph "Private Subnet A (Application)"
            CmdSvc[Command Service]
            QuerySvc[Query Service]
        end

        subgraph "Private Subnet B (Data Pipeline)"
            Kafka[Kafka Cluster]
            Spark[Spark Cluster]
        end

        subgraph "Private Subnet C (Data Stores)"
            Cassandra[Cassandra Cluster]
            RecoilApp[Frontend serving from S3/CloudFront]
        end

        Internet --> ALB
        ALB -- SG_App --> CmdSvc
        ALB -- SG_App --> QuerySvc

        CmdSvc -- SG_Kafka_Producer --> Kafka
        Spark -- SG_Kafka_Consumer --> Kafka
        Spark -- SG_Cassandra_Writer --> Cassandra

        QuerySvc -- SG_Cassandra_Reader --> Cassandra
    end

    RecoilApp -- API Calls --> ALB

VPC：提供了一個邏輯上隔離的網絡空間。
子網：
- Public Subnet：放置面向公網的負載均衡器（ALB）或 API Gateway，作為流量入口。
- Private Subnets：所有核心應用和數據組件都部署在私有子網中，無法從公網直接訪問，確保安全。
**安全組 (Security Groups)**：作為實例級別的防火牆，精確控制組件間的流量。例如，只有 Command Service 的安全組被允許向 Kafka 的端口發送數據；只有 Spark 集群的安全組被允許從 Kafka 消費數據並寫入 Cassandra。這是實現最小權限原則的關鍵。

3.2. 事件生成與持久化 (Command Service)

這是一個簡單的 Node.js 服務，使用 KafkaJS 客戶端。它的職責是接收 HTTP 請求，將其轉化為領域事件，並發送到 Kafka。

// command-service/src/handler.js
const { Kafka, CompressionTypes } = require('kafkajs');
const crypto = require('crypto');

// 在生產環境中，這些配置應來自環境變量或配置中心
const kafka = new Kafka({
  clientId: 'command-service',
  brokers: ['kafka-broker1:9092', 'kafka-broker2:9092'],
});

const producer = kafka.producer();
const TOPIC_NAME = 'user-interaction-events';

// 確保在服務啟動時連接，在關閉時斷開
const connectProducer = async () => {
  try {
    await producer.connect();
    console.log('Kafka producer connected successfully.');
  } catch (error) {
    console.error('Failed to connect Kafka producer:', error);
    // 實現重試邏輯或直接退出進程
    process.exit(1);
  }
};

connectProducer();

// 假設這是一個 Express.js 的路由處理器
async function handleUserClick(req, res) {
  const { userId, elementId, timestamp } = req.body;

  // 業務驗證邏輯
  if (!userId || !elementId) {
    return res.status(400).json({ error: 'Missing required fields' });
  }

  const event = {
    eventId: crypto.randomUUID(),
    eventType: 'UserClickedElement',
    eventTimestamp: new Date().toISOString(),
    payload: {
      userId,
      elementId,
      // 確保傳入的時間戳是有效的，否則使用服務器時間
      clientTimestamp: timestamp || new Date(timestamp).toISOString(), 
    },
    schemaVersion: '1.0.0' // 事件模式版本管理至關重要
  };

  try {
    // 使用 userId 作為 key，保證同一個用戶的事件進入同一個 Kafka 分區，維持順序性
    await producer.send({
      topic: TOPIC_NAME,
      compression: CompressionTypes.GZIP, // 對消息進行壓縮，節省帶寬和存儲
      messages: [
        { key: userId, value: JSON.stringify(event) },
      ],
    });

    // 重要的不是返回數據，而是確認事件已被接收
    res.status(202).json({ message: 'Event accepted', eventId: event.eventId });

  } catch (error) {
    console.error(`[${event.eventId}] Failed to send event to Kafka:`, error);
    // 實現錯誤處理，例如將失敗的事件存入死信隊列
    res.status(500).json({ error: 'Internal server error while processing event' });
  }
}

// 服務關閉時的優雅處理
process.on('SIGTERM', async () => {
  console.log('SIGTERM received, disconnecting producer...');
  await producer.disconnect();
  process.exit(0);
});

關鍵考量：

冪等性：eventId 的存在是為了讓下游消費者具備處理重複消息的能力。
順序保證：將 userId 作為 Kafka 消息的 key，確保了單個用戶的事件在分區內是有序的，這對於很多狀態計算至關重要。
錯誤處理：生產級代碼必須考慮 Kafka broker 不可用或消息發送失敗的情況，並設計相應的重試或降級策略。

3.3. 流式處理與投影 (Apache Spark)

Apache Spark Streaming 作業負責實時消費 Kafka 中的事件，進行有狀態的計算（例如，計算每個元素每分鐘的點擊次數），並將結果寫入 Cassandra。

// spark-projector/src/main/scala/com/example/analytics/EventProcessor.scala
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.{StringType, StructType, TimestampType}

// 引入 Cassandra 連接器
import com.datastax.spark.connector.cql.CassandraConnectorConf
import com.datastax.spark.connector.writer.WriteConf

object EventProcessor {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("RealtimeEventProjector")
      // 生產環境中 master URL 會由 spark-submit 指定
      .master("local[*]") 
      // Cassandra 連接配置
      .config("spark.cassandra.connection.host", "cassandra-node1,cassandra-node2")
      .config("spark.cassandra.connection.port", "9042")
      .config("spark.cassandra.auth.username", "cassandra")
      .config("spark.cassandra.auth.password", "password")
      .getOrCreate()

    spark.sparkContext.setLogLevel("WARN")

    import spark.implicits._

    // 定義事件的 Schema
    val eventSchema = new StructType()
      .add("eventId", StringType)
      .add("eventType", StringType)
      .add("eventTimestamp", TimestampType)
      .add("payload", new StructType()
        .add("userId", StringType)
        .add("elementId", StringType)
        .add("clientTimestamp", StringType))
      .add("schemaVersion", StringType)

    // 從 Kafka 中讀取數據流
    val kafkaStreamDF = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092")
      .option("subscribe", "user-interaction-events")
      .option("startingOffsets", "latest") // 從最新的消息開始消費
      .option("failOnDataLoss", "false") // 生產環境需要仔細評估此選項
      .load()

    val eventsDF = kafkaStreamDF
      .selectExpr("CAST(value AS STRING) as json")
      .select(from_json($"json", eventSchema).as("data"))
      .select("data.*")
      
    // 核心業務邏輯：按窗口計算點擊次數
    val windowedCounts = eventsDF
      .withWatermark("eventTimestamp", "10 minutes") // 處理延遲數據
      .groupBy(
        window($"eventTimestamp", "1 minute", "1 minute"), // 1 分鐘的滾動窗口
        $"payload.elementId"
      )
      .count()
      .select(
        $"window.start".as("window_start"),
        $"elementId",
        $"count"
      )

    // 將結果寫入 Cassandra
    val query = windowedCounts.writeStream
      .foreachBatch { (batchDF, batchId) =>
        println(s"Processing batch $batchId...")
        batchDF.write
          .format("org.apache.spark.sql.cassandra")
          .options(Map("table" -> "element_clicks_per_minute", "keyspace" -> "analytics"))
          .mode("append")
          .save()
      }
      .outputMode("update") // 只輸出更新的行
      .trigger(Trigger.ProcessingTime("30 seconds")) // 每 30 秒觸發一次計算
      .option("checkpointLocation", "/path/to/spark/checkpoint") // Exactly-once 語義的保障
      .start()

    query.awaitTermination()
  }
}

關鍵考量：

Checkpointing：checkpointLocation 是實現端到端 Exactly-once 語義的關鍵。Spark 會將 offset 和計算狀態保存到這裡，確保在作業失敗重啟後能從斷點處恢復，不丟失不重複。
Watermarking：withWatermark 用於處理事件時間（event time）和處理時間（processing time）的偏差，能夠容忍一定程度的數據延遲到達。
Sink 冪等性：Cassandra 的 upsert 操作是冪等的，這意味著即使 Spark 因為重試而多次寫入相同的數據，最終結果也是正確的，這對保證數據準確性至關重要。

3.4. 前端狀態管理與展示 (Recoil)

前端使用 React 和 Recoil 來構建儀表盤。Recoil 的 selector 非常適合用來從 Query Service 獲取數據並進行派生計算。

// dashboard/src/state/analyticsState.js
import { atom, selector } from 'recoil';
import axios from 'axios';

// 一個 atom 用於控制當前查看的元素 ID
export const selectedElementIdState = atom({
  key: 'selectedElementId',
  default: 'cta-button-main',
});

// selector 異步獲取後端讀模型的數據
export const elementClickStatsSelector = selector({
  key: 'elementClickStats',
  get: async ({ get }) => {
    const elementId = get(selectedElementIdState);
    if (!elementId) return null;

    try {
      // 實際項目中 API 地址應通過配置管理
      const response = await axios.get(`/api/analytics/clicks?elementId=${elementId}`);
      // 在此可以對數據進行一些前端的轉換或清理
      return response.data;
    } catch (error) {
      console.error(`Failed to fetch stats for ${elementId}:`, error);
      // Recoil 的 selector 支持拋出異常，可以被 React.Suspense 和 ErrorBoundary 捕獲
      throw error;
    }
  },
});

Recoil 的原子化狀態管理模型，使得UI組件可以僅訂閱它們關心的那一小片數據。當 selectedElementIdState 變化時，只有 elementClickStatsSelector 和使用它的組件會重新計算和渲染，非常高效。

3.5. 端到端測試驗證 (Cypress)

測試這個異步、分佈式的系統是最大的挑戰。我們需要編寫一個 Cypress E2E 測試，它能觸發一個動作，然後輪詢驗證整個數據管道是否按預期工作，最終反映在 UI 上。

// cypress/e2e/analytics-flow.cy.js
describe('Real-time Analytics Flow E2E Test', () => {
  
  // Cypress task 可以執行後端 Node.js 代碼，用於數據準備或與後端直接交互
  // 在 cypress.config.js 中需要配置 task
  it('should reflect a new click event on the dashboard after processing', () => {
    const uniqueElementId = `test-element-${Date.now()}`;
    const userId = 'test-user-e2e';
    const initialClickCount = 0; // 假設是新元素

    // Step 1: 訪問頁面並確認初始狀態
    cy.visit('/dashboard');
    cy.get(`[data-testid="stats-for-${uniqueElementId}"]`).should('not.exist');
    
    // Step 2: 直接調用 API 發送一個命令，模擬用戶點擊
    // 這種方式比模擬 UI 點擊更穩定，能將前後端測試解耦
    cy.request('POST', 'http://localhost:3000/api/events/click', {
      userId: userId,
      elementId: uniqueElementId,
      timestamp: new Date().toISOString()
    }).its('status').should('eq', 202);

    // Step 3: 輪詢讀模型 API，等待 Spark 處理完事件
    // 這是測試最終一致性系統的關鍵。嚴禁使用 cy.wait(fixed_time)
    const pollOptions = {
      timeout: 30000, // 最長等待 30 秒
      interval: 2000, // 每 2 秒輪詢一次
      errorMsg: 'Analytics data was not updated in time',
    };

    cy.log('Polling read model for updated stats...');
    cy.waitUntil(() => 
      cy.request({
        url: `http://localhost:3001/api/analytics/clicks?elementId=${uniqueElementId}`,
        failOnStatusCode: false // 允許 초기 404
      }).then(response => {
        if (response.status === 200 && response.body && response.body.totalClicks > initialClickCount) {
          cy.log('Data found in read model!');
          return true; // 條件滿足，停止輪詢
        }
        cy.log('Data not yet available, retrying...');
        return false;
      }), pollOptions);

    // Step 4: 刷新頁面，驗證 UI 是否正確展示了新數據
    cy.visit('/dashboard'); // 重新加載以獲取最新數據
    cy.get(`[data-testid="stats-for-${uniqueElementId}"] [data-testid="click-count"]`)
      .should('contain.text', initialClickCount + 1);
  });
});

關鍵測試策略：

API 驅動：測試直接通過 API 發送命令，而不是模擬 UI 點擊。這使得測試更快速、更穩定，且不受 UI 變化的影響。
智能輪詢：使用 waitUntil 這樣的自定義命令或庫（例如 cypress-wait-until）來輪詢後端讀模型，直到數據就緒。這是驗證最終一致性系統的唯一可靠方法。
隔離測試數據：使用時間戳或隨機字符串生成唯一的 elementId，確保測試之間互不干擾。

4. 架構的局限性與未來展望

儘管這套基於事件溯源和 Spark 的 CQRS 架構非常強大，但它並非銀彈。

延遲問題：從事件產生到讀模型可用，整個鏈路存在固有的延遲（Kafka 傳輸、Spark 批處理間隔等）。對於需要亞秒級讀寫一致性的場景，此架構可能不適用。需要仔細定義和監控端到端的數據延遲 SLO。
運維複雜度：管理一個分佈式的系統，包括 VPC、Kafka、Spark、Cassandra，需要一個成熟的 SRE 或 DevOps 團隊以及強大的可觀測性平台（日誌、指標、追蹤）。
事件模式演進：一旦事件被寫入事件存儲，它們就是不可變的。如果事件的結構需要改變（例如，增加一個字段），就需要一套完整的版本控制和遷移策略，以確保新舊版本的消費者都能正確處理。這通常是事件溯源中最棘手的部分。

未來的優化方向可能包括引入 Flink 以獲得更低的流處理延遲，或者探索如 Apache Pinot、ClickHouse 等專用 OLAP 引擎作為讀模型存儲，以提供更強大的實時分析查詢能力。同時，前端可以引入 WebSocket，將讀模型的更新實時推送給客戶端，進一步提升用戶體驗。

Event Sourcing Apache Spark Cypress Recoil VPC

构建由 Apache Spark 驱动并基于 Qwik 与 PostgreSQL 的大规模机器学习特征可观测性平台

2023-10-27 数据工程

Qwik Vite Apollo Client Apache Spark PostgreSQL

基于非阻塞IO在OpenFaaS中构建高性能C++数据库写入函数

2023-10-27 Serverless

性能优化 PostgreSQL Database OpenFaaS C++