Tất tần tật về Change Data Capture và Debezium (Phần 1)

Mỗi hệ thống khi vận hành sẽ đều có những dữ liệu, những transaction được lưu trữ trong operational database. Những transaction này có ý nghĩa rất lớn và rất quan trọng vì qua đó, ta sẽ hiểu được insight khách hàng, kết quả kinh doanh, trạng thái vận hành của doanh nghiệp… Và thông thường, để tránh ảnh hướng để hiệu suất của operational system, dữ liệu transaction sẽ được nhân bản (replicate) sang Data Warehouse cho mục đích phân tích.

Một cách truyền thống, các hệ thống thường xử lý dữ liệu từng batch (batch-processing) để chuyển dữ liệu sang Data Warehouse một hoặc vài lần trong ngày. Tuy nhiên cách làm này sẽ tạo ra độ trễ và giảm giá trị của dữ liệu. Với một số business cụ thể, giá trị của dữ liệu sẽ giảm dần theo thời gian, ví dụ như dữ liệu tài chính, cổ phiếu…

Từ đây, nhu cầu về một phương pháp để migrate và replicate dữ liệu liên tục giữa những database là rất cấp thiết. Change Data Capture đã xuất hiện như một giải pháp để có thể liên tục phát hiện thay đổi ở source database, từ đó giúp cho hệ thống có thể xử lý và phân tích dữ liệu ngay lập tức.

Change Data Capture là gì?

Change Data Capture (CDC) là một kỹ thuật cũng như một kiểu kiến trúc để thiết kế hệ thống mà qua đó ta có thể theo dõi được những thay đổi phát sinh ở phía source database.

Với việc sử dụng CDC, ta có thể dễ dàng sao chép, nhân bản (replicate) hoặc chuyển đổi (migrate) dữ liệu giữa nhiều database khác nhau trong thời gian thực (real-time).

Nhờ CDC, hệ thống sẽ có thể từ từ load được những thay đối ở phía source database (incremental load) thay vì phải load một lượng lớn dữ liệu theo từng đợt (bulk load).

Ưu điểm của CDC

So với lối xử lý dữ liệu truyền thống batch processing, CDC đem đến rất nhiều ưu điểm rõ rệt:

Debezium

Ở phần này, chúng ta sẽ tập trung tìm hiểu về một công cụ để làm CDC, đó là Debezium.

Debezium bao gồm 1 tập hợp các dịch vụ phân tán (distributed services) để theo dõi và bắt được những thay đổi diễn ra ở phía source database.

Thông thường, mỗi một transaction xảy ra ở database đều được lưu lại tại transaction log và các database khác nhau sẽ có cách ghi transaction log khác nhau.

Debezium sẽ sử dụng những connector đọc transaction log để phát hiện những thay đổi đã xảy ra với source database, nhờ đó mà Debezium có thể phát hiện thay đổi với từng dòng (row-level), từng record. Sau đó, những change-event tương ứng với từng transaction sẽ được tạo ra và gửi đến những streaming service như Kafka, AWS Kinesis, …

Đến thời điểm hiện tại, Detạibezium đã hỗ trợ cả Relational và Non-Relational Database, ví dụ như MySQL, SQL Server, MongoDB…

Embedded Debezium

Chúng ta sẽ cùng tìm hiểu Embedded Debezium, một phiên bản Debezium với kiến trúc nhỏ gọn và đơn giản nhất.

Thông thường, Debezium được vận hành bằng cách deploy lên Kafka Connect, một hoặc nhiều connector sẽ được sử dụng để theo dõi source database và tạo ra những change-event cho từng thay đổi ở source database. Những change-event này sẽ được ghi và lưu lên Kafka, sau đó các ứng dụng phía target có thể lấy ra (consume) những change-event này từ Kafka.

Về phía Kafka Connect, ưu điểm của nó là đem đến khả năng chịu lỗi (fault tolerance) và mở rộng (scalability) xuất sắc. Nó sẽ hoạt động như một distributed service và đảm bảo rằng các connector đã đăng ký sẽ luôn chạy ổn định. Ví dụ, ngay cả khi một endpoints của Kafka Connect không hoạt động, những endpoints còn lại sẽ tự động khởi động lại các connector trước đó đang chạy trên endpoints cũ, nhờ đó giảm tối đa thời gian chết (minimize downtime) của hệ thống.

Tuy nhiên, không phải ứng dụng nào cũng cần mức độ chịu lỗi quá cao, nhiều ứng dụng còn không muốn phụ thuộc vào cluster bên ngoài như Kafka và Kafka Connect và ưa thích một hệ thống nhỏ gọn. Vì vậy, thay vì dùng Kafka Connect, những app kiểu này sẽ tích hợp trực tiếp Debezium connector vào trong app. Do đó, Debezium đã cung cấp một module gọi là debezium-api, cho phép ứng dụng có thể dễ dàng cấu hình và sử dụng Debezium connector và Debezium Engine.

Đây là kiến trúc mẫu khi sử dụng phiên bản Embedded Debezium trong ứng dụng Java:

Trong kiến trúc mẫu trên, ta có 1 source database, 1 app Java Spring Boot, 1 target database.

Bên trong app Java ta sẽ tích hợp Debezium Engine và viết sẵn những function để có thể theo dõi và bắt được thay đổi ở phía source database mình muốn.

Sau đó, những change-event mà Debezium phát hiện được từ phía source database cũng có thể được xử lý luôn trong ứng dụng Java này (Ví dụ: Ta có thể trích xuất thông tin từ change-event, sau đó thực hiện update vào target database).

Ví dụ: Maven dependencies được cấu hình thông qua config file.

<dependency>

    <groupId>io.debezium</groupId>

    <artifactId>debezium-api</artifactId>

    <version>1.4.2.Final</version>

</dependency>

<dependency>

    <groupId>io.debezium</groupId>

    <artifactId>debezium-embedded</artifactId>

    <version>1.4.2.Final</version>

</dependency>

<dependency>

    <groupId>io.debezium</groupId>

    <artifactId>debezium-connector-mysql</artifactId>

    <version>1.4.2.Final</version>

</dependency>

Đây là một đoạn sample code Java có sử dụng Embedded Debezium:

// Define the configuration for the Debezium Engine with MySQL connector...
final Properties props = config.asProperties();
props.setProperty("name", "engine");
props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
props.setProperty("offset.storage.file.filename", "/tmp/offsets.dat");
props.setProperty("offset.flush.interval.ms", "60000");
/* begin connector properties */
props.setProperty("database.hostname", "localhost");
props.setProperty("database.port", "3306");
props.setProperty("database.user", "mysqluser");
props.setProperty("database.password", "mysqlpw");
props.setProperty("database.server.id", "85744");
props.setProperty("database.server.name", "my-app-connector");
props.setProperty("database.history",
      "io.debezium.relational.history.FileDatabaseHistory");
props.setProperty("database.history.file.filename",
      "/path/to/storage/dbhistory.dat");

// Create the engine with this configuration ...
try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
        .using(props)
        .notifying(record -> {
            System.out.println(record);
        }).build()
    ) {
    // Run the engine asynchronously ...
    ExecutorService executor = Executors.newSingleThreadExecutor();
    executor.execute(engine);

    // Do something else or wait for a signal or an event
}
// Engine is stopped when the main code is finished

Đọc thêm chi tiết về Debezium Engine tại đây.

Kiến trúc khi sử dụng với Kafka Connect

Kiến trúc phổ biến nhất của một hệ thống sử dụng Debezium chính là tích hợp Debezium cùng với Kafka Connect.

Nếu bạn chưa biết về Kafka Connect thì đây là một framework hoạt động như một service riêng biệt và chạy song song với Kafka Broker, mục đích để truyền dữ liệu từ các hệ thống khác đến Kafka và ngược lại (truyền dữ liệu từ Kafka ra các hệ thống khác).

Với kiến trúc như trên, Kafka Connect sẽ đóng vai trò quản lý và vận hành các module như:

• Source Connector: Những connector như Debezium connector dùng để theo dõi source database
• Sink Connector: Connector để truyền dữ liệu từ Kafka đến target database

Trong kiến trúc mẫu bên trên, ta sẽ có:

• 2 source database cần được theo dõi - MySQL DB và PostgreSQL DB
• 2 Debezium connector tương ứng với 2 database
• 2 Connector này sẽ được đăng ký với Kafka Connect, từ đó Kafka Connect sẽ quản lý và truyền dữ liệu từ Debezium đến Kafka Topic
• 2 Kafka Connect: 1 để quản lý source connector, 1 để quản lý sink connector
• Kafka: Streaming Service chứa event nhận được từ Debezium
• 3 sink connector: Elastic Search connector, Infinispan connector, JDBC connector để truyền dữ liệu từ Kafka đến các system bên ngoài
• 3 target system: Elastic Search, Infinispan, Data Warehouse

Debezium Server

Ngoài tích hợp với Kafka Connect, Debezium cũng có thể tích hợp với Debezium Server như một giải pháp thay thế để truyền event đến các streaming service như: Amazon Kinesis, Google Pub/Sub, Pulsar…

Nếu bạn chưa biết thì Debezium Server là một ứng dụng hỗ trợ truyền change-event từ Debezium đến các service xử lý message (message infrastructure). Khi tích hợp Debezium với Debezium Server, ta có thể setup cấu hình qua file config một cách dễ dàng.

Đọc thêm chi tiết tại đây.

Đánh giá hiệu năng

Sau khi đã hiểu về tác dụng của Debezium và các kiến trúc phổ biến của hệ thống tích hợp Debezium, ta sẽ đến một phần rất quan trọng, đó là đánh giá hiệu năng của công cụ này.

Bài test này được thực hiện trên một máy Azure VM với cấu hình như sau:

Ta sẽ đánh giá hiệu năng của Debezium trong 2 kịch bản:

1. Trường hợp 1

Mô tả:

• Phía source database đang có rất nhiều dữ liệu.
• Phía target database chưa có record nào.
• Debezium connector sau khi được đăng ký (register) với Kafka Connect thì bắt đầu thực hiện initial snapshot để chụp lại tất cả những dữ liệu mà phía source database đang có.
• Sau đó tất cả những record của phiên bản snapshot này đều được tính là những record mới với phía target database và các change-event tương ứng được tạo ra.
• Quá trình đồng bộ giữa source database và target database cũng bắt đầu.

Thông số:

2. Trường hợp 2

Mô tả:

• Sau khi 2 bên database (source và target) đã có số lượng dữ liệu bằng nhau và dữ liệu cũng giống như nhau, phía source database tự nhiên phát sinh một lượng lớn update.
• Có khoảng gần 10 triệu record được update.
• Chúng ta sẽ đo xem sau bao lâu thì dữ liệu của source và target database được đồng bộ hoàn toàn.

Thông số:

Ưu và nhược điểm

Trên đời không có gì là hoàn hảo và Debezium cũng thế. Công cụ này có cả những ưu điểm và nhược điểm của riêng mình.

Debezium khá gọn nhẹ, chỉ tập trung vào chức năng CDC. Ngoài ra công cụ này còn có Apache License nên sử dụng hoàn toàn miễn phí.

Vậy là chúng ta đã kết thúc phần 1 của series tìm hiểu về Change Data Capture và Debezium. Rất cảm ơn bạn đã theo dõi đến đây. Trong phần tiếp theo, chúng ta sẽ có những demo cụ thể để hiểu sâu hơn về cách sử dụng Debezium.