DMS-CDC-Operator

概述

rust-dms-cdc-operator 是一个基于Rust的工具，用于比较Amazon RDS（PostgreSQL）数据库中的表与使用AWS数据库迁移服务（DMS）迁移到Amazon S3的数据。它特别有助于确保RDS数据库与S3中的Parquet文件之间的数据一致性，特别是在数据捕获（CDC）更新方面，因为目前还不支持以S3为目标进行DMS验证。

特性

• 将存储在AWS S3中的基于日期文件夹分区的CDC parquet数据快照导入本地部署的Postgres
• 指定特定的时间范围以在Postgres数据库上复制S3状态
• 在数据丢失的情况下从S3恢复RDS状态
• 将Amazon RDS数据库中特定表的状态与S3存储桶中存储的Parquet文件中的数据进行比较
• 通过修改验证的块大小来识别行级差异
• 将其作为库使用，以便将其集成到项目中，或作为客户端使用，以便作为独立工具使用

先决条件

• 您的源数据库是PostgreSQL
• 您有一个运行在完全加载（或完全加载 + CDC）模式的AWS DMS任务
• 任务的目的是AWS S3
  • Parquet格式的文件
  • 基于日期的文件夹分区
  • DMS注入的额外列 Op

安装（客户端）

为了将工具作为客户端使用，您可以使用 cargo。

该工具提供了两个用于运行它的功能，分别是 Inquire 和 Clap。

使用Clap

Usage: dms-cdc-operator-client validate [OPTIONS] --bucket-name <BUCKET_NAME> --s3-prefix <S3_PREFIX> --source-postgres-url <SOURCE_POSTGRES_URL> --target-postgres-url <TARGET_POSTGRES_URL>

Options:
      --bucket-name <BUCKET_NAME>
          S3 Bucket name where the CDC files are stored
      --s3-prefix <S3_PREFIX>
          S3 Prefix where the files are stored Example: data/landing/rds/mydb
      --source-postgres-url <SOURCE_POSTGRES_URL>
          Url of the database to validate the CDC files Example: postgres://postgres:postgres@localhost:5432/mydb
      --target-postgres-url <TARGET_POSTGRES_URL>
          Url of the target database to import the parquet files Example: postgres://postgres:postgres@localhost:5432/mydb
      --database-schema <DATABASE_SCHEMA>
          Schema of database to validate against S3 files [default: public]
      --included-tables [<INCLUDED_TABLES>...]
          List of tables to include for validatation against S3 files
      --excluded-tables [<EXCLUDED_TABLES>...]
          List of tables to exclude for validatation against S3 files
      --mode <MODE>
          Mode to load Parquet files Example: DateAware Example: AbsolutePath Example: FullLoadOnly [default: date-aware] [possible values: date-aware, absolute-path, full-load-only]
      --start-date <START_DATE>
          Start date to filter the Parquet files Example: 2024-02-14T10:00:00Z
      --stop-date <STOP_DATE>
          Stop date to filter the Parquet files Example: 2024-02-14T10:00:00Z
      --chunk-size <CHUNK_SIZE>
          Datadiff chunk size [default: 1000]
      --max-connections <MAX_CONNECTIONS>
          Maximum connection pool size [default: 100]
      --start-position <START_POSITION>
          Datadiff start position [default: 0]
      --only-datadiff
          Run only the datadiff
      --only-snapshot
          Take only a snapshot from S3 to target DB
      --accept-invalid-certs-first-db
          Accept invalid TLS certificates for the first database
      --accept-invalid-certs-second-db
          Accept invalid TLS certificates for the second database
  -h, --help
          Print help
  -V, --version
          Print version

使用Inquire

rust-cdc-validator --features="with-inquire"

安装（库）

为了将工具作为库使用，您可以运行

cargo add rust-cdc-validator

示例

• 通过Docker Compose启动本地Postgres数据库

docker-compose up

psql -h localhost -p 5438 -U postgres -d mydb

• 构建和运行Rust工具

cargo fmt --all
cargo clippy --all

cargo build

RUST_LOG=dms_cdc_operator=info,rust_pgdatadiff=info cargo run --features="with-clap" validate --bucket-name my-bucket --s3-prefix prefix/path --source-postgres-url postgres://postgres:postgres@localhost:5432/mydb1 --target-postgres-url postgres://postgres:postgres@localhost:5438/mydb --database-schema public --included-tables mytable --start-date 2024-02-14T10:00:00Z --chunk-size 100

为了进行更多调试，您可以通过导出以下内容来启用与Rust相关的日志

export RUST_LOG=dms_cdc_operator=debug,rust_pgdatadiff=debug

许可证

本项目采用MIT许可证