同步数据到存储服务

从 TiDB v6.5.0 开始,TiCDC 支持将行变更事件保存至存储服务,如 Amazon S3、GCS、Azure Blob Storage 和 NFS。本文介绍如何使用 TiCDC 创建同步任务 (Changefeed) 将增量数据同步到这类存储服务,并介绍数据的存储方式。具体如下:

同步变更数据至存储服务

使用以下命令来创建同步任务:

cdc cli changefeed create \ --server=http://10.0.10.25:8300 \ --sink-uri="s3://logbucket/storage_test?protocol=canal-json" \ --changefeed-id="simple-replication-task"

输出结果如下:

Info: {"upstream_id":7171388873935111376,"namespace":"default","id":"simple-replication-task","sink_uri":"s3://logbucket/storage_test?protocol=canal-json","create_time":"2024-01-25T18:52:05.566016967+08:00","start_ts":437706850431664129,"engine":"unified","config":{"case_sensitive":false,"enable_old_value":true,"force_replicate":false,"ignore_ineligible_table":false,"check_gc_safe_point":true,"enable_sync_point":false,"sync_point_interval":600000000000,"sync_point_retention":86400000000000,"filter":{"rules":["*.*"],"event_filters":null},"mounter":{"worker_num":16},"sink":{"protocol":"canal-json","schema_registry":"","csv":{"delimiter":",","quote":"\"","null":"\\N","include_commit_ts":false},"column_selectors":null,"transaction_atomicity":"none","encoder_concurrency":16,"terminator":"\r\n","date_separator":"none","enable_partition_separator":false},"consistent":{"level":"none","max_log_size":64,"flush_interval":2000,"storage":""}},"state":"normal","creator_version":"v7.6.0"}
  • --server:TiCDC 集群中任意一个 TiCDC 服务器的地址。
  • --changefeed-id:同步任务的 ID。格式需要符合正则表达式 ^[a-zA-Z0-9]+(\-[a-zA-Z0-9]+)*$。如果不指定该 ID,TiCDC 会自动生成一个 UUID(version 4 格式)作为 ID。
  • --sink-uri:同步任务下游的地址。具体可参考配置 Sink URI
  • --start-ts:指定 changefeed 的开始 TSO。TiCDC 集群将从这个 TSO 开始拉取数据。默认为当前时间。
  • --target-ts:指定 changefeed 的目标 TSO。TiCDC 集群拉取数据直到这个 TSO 停止。默认为空,即 TiCDC 不会自动停止。
  • --config:指定 changefeed 配置文件,详见 TiCDC Changefeed 配置参数

配置 Sink URI

本章节介绍如何在 Sink URI 中配置存储服务 Amazon S3、GCS、Azure Blob Storage 以及 NFS。Sink URI 用于指定 TiCDC 下游系统的连接信息,遵循以下格式:

[scheme]://[host]/[path]?[query_parameters]

URI 的 [query_parameters] 中可配置的参数如下:

参数描述默认值取值范围
worker-count向下游存储服务保存数据变更记录的并发度16[1, 512]
flush-interval向下游存储服务保存数据变更记录的间隔5s[2s, 10m]
file-size单个数据变更文件的字节数超过 file-size 时将其保存至存储服务中67108864[1048576, 536870912]
protocol输出到存储服务的消息协议N/Acanal-jsoncsv
enable-tidb-extensionprotocol 参数为 canal-json 时,如果该值为 true,TiCDC 会发送 WATERMARK 事件,并在 canal-json 消息中添加 TiDB 扩展字段falsefalsetrue

配置外部存储

Amazon S3 配置样例如下:

--sink-uri="s3://bucket/prefix?protocol=canal-json"

GCS 配置样例如下:

--sink-uri="gcs://bucket/prefix?protocol=canal-json"

Azure Blob Storage 配置样例如下:

--sink-uri="azure://bucket/prefix?protocol=canal-json"

配置 NFS

NFS 配置样例如下:

--sink-uri="file:///my-directory/prefix?protocol=canal-json"

存储路径组织结构

本章节详细介绍数据变更记录、元数据与 DDL 事件的存储路径组织结构。

数据变更记录

数据变更记录将会存储到以下路径:

{scheme}://{prefix}/{schema}/{table}/{table-version-separator}/{partition-separator}/{date-separator}/CDC{num}.{extension}
  • scheme:存储服务类型。例如:s3gcsazurefile
  • prefix:用户指定的父目录。例如:s3://bucket/bbb/ccc
  • schema:表所属的库名。例如:s3://bucket/bbb/ccc/test
  • table:表名。例如:s3://bucket/bbb/ccc/test/table1
  • table-version-separator:将文件路径按照表的版本进行分隔。例如:s3://bucket/bbb/ccc/test/table1/9999
  • partition-separator:将文件路径按照表的分区号进行分隔。例如:s3://bucket/bbb/ccc/test/table1/9999/20
  • date-separator:将文件路径按照事务提交的日期进行分隔,默认值为 day,可选值如下:
    • none:不以 date-separator 分隔文件路径。例如:test.table1 版本号为 9999 的所有文件都存到 s3://bucket/bbb/ccc/test/table1/9999 路径下。
    • year:以事务提交的年份分隔文件路径。例如:s3://bucket/bbb/ccc/test/table1/9999/2022
    • month:以事务提交的年份和月份分隔文件路径。例如:s3://bucket/bbb/ccc/test/table1/9999/2022-01
    • day:以事务提交的年月日来分隔文件路径。例如:s3://bucket/bbb/ccc/test/table1/9999/2022-01-02
  • num:存储数据变更记录的目录下文件的序号。例如:s3://bucket/bbb/ccc/test/table1/9999/2022-01-02/CDC000005.csv
  • extension:文件的扩展名。v6.5.0 支持 CSV 和 Canal-JSON 格式。

Index 文件

Index 文件用于防止已写入的数据被错误覆盖,与数据变更记录存储在相同路径:

{scheme}://{prefix}/{schema}/{table}/{table-version-separator}/{partition-separator}/{date-separator}/meta/CDC.index

Index 文件记录了当前目录下所使用到的最大文件名,比如:

CDC000005.csv

上述内容表明该目录下 CDC000001.csvCDC000004.csv 文件已被占用,当 TiCDC 集群中发生表调度或者节点重启时,新的节点会读取 Index 文件,并判断 CDC000005.csv 是否被占用。如果未被占用,则新节点会从 CDC000005.csv 开始写文件。如果已被占用,则从 CDC000006.csv 开始写文件,这样可防止覆盖其他节点写入的数据。

元数据

元数据信息将会存储到以下路径:

{scheme}://{prefix}/metadata

元数据信息以 JSON 格式存储到如下的文件中:

{ "checkpoint-ts":433305438660591626 }
  • checkpoint-ts:commit-ts 小于等于此 checkpoint-ts 的事务都被写入下游存储当中。

DDL 事件

表级 DDL 事件

当上游表的 DDL 事件引起表的版本变更时,TiCDC 将会自动进行以下操作:

  • 切换到新的路径下写入数据变更记录。例如,当 test.table1 的版本变更为 441349361156227074 时,TiCDC 将会在 s3://bucket/bbb/ccc/test/table1/441349361156227074/2022-01-02/ 路径下写入数据。

  • 生成一个 schema 文件存储表结构信息,文件路径如下:

    {scheme}://{prefix}/{schema}/{table}/meta/schema_{table-version}_{hash}.json

schema_441349361156227074_3131721815.json 为例,表结构信息文件的内容如下:

{ "Table":"table1", "Schema":"test", "Version":1, "TableVersion":441349361156227074, "Query":"ALTER TABLE test.table1 ADD OfficeLocation blob(20)", "Type":5, "TableColumns":[ { "ColumnName":"Id", "ColumnType":"INT", "ColumnNullable":"false", "ColumnIsPk":"true" }, { "ColumnName":"LastName", "ColumnType":"CHAR", "ColumnLength":"20" }, { "ColumnName":"FirstName", "ColumnType":"VARCHAR", "ColumnLength":"30" }, { "ColumnName":"HireDate", "ColumnType":"DATETIME" }, { "ColumnName":"OfficeLocation", "ColumnType":"BLOB", "ColumnLength":"20" } ], "TableColumnsTotal":"5" }
  • Table:表名。
  • Schema:表所属的库名。
  • Version:Storage sink 协议版本号。
  • TableVersion:表的版本号。
  • Query:DDL 语句。
  • Type:DDL 类型。
  • TableColumns:该数组表示表中每一列的详细信息。
    • ColumnName:列名。
    • ColumnType:该列的类型。详见数据类型
    • ColumnLength:该列的长度。详见数据类型
    • ColumnPrecision:该列的精度。详见数据类型
    • ColumnScale:该列小数位的长度。详见数据类型
    • ColumnNullable:值为 true 时表示该列可以含 NULL 值。
    • ColumnIsPk:值为 true 时表示该列是主键的一部分。
  • TableColumnsTotalTableColumns 数组的大小。

库级 DDL 事件

当上游数据库发生库级 DDL 事件时,TiCDC 将会自动生成一个 schema 文件存储数据库结构信息,文件路径如下:

{scheme}://{prefix}/{schema}/meta/schema_{table-version}_{hash}.json

schema_441349361156227000_3131721815.json 为例,数据库结构信息文件的内容如下:

{ "Table": "", "Schema": "schema1", "Version": 1, "TableVersion": 441349361156227000, "Query": "CREATE DATABASE `schema1`", "Type": 1, "TableColumns": null, "TableColumnsTotal": 0 }

数据类型

本章节主要介绍 schema_{table-version}_{hash}.json 文件(以下简称为 schema 文件)中使用的各种数据类型。数据类型定义为 T(M[, D]),详见数据类型概述

整数类型

TiDB 中整数类型可被定义为 IT[(M)] [UNSIGNED],其中:

  • IT 为整数类型,包括 TINYINTSMALLINTMEDIUMINTINTBIGINTBIT
  • M 为该类型的显示宽度。

schema 文件中对整数类型定义如下:

{ "ColumnName":"COL1", "ColumnType":"{IT} [UNSIGNED]", "ColumnPrecision":"{M}" }

小数类型

TiDB 中的小数类型可被定义为 DT[(M,D)][UNSIGNED],其中:

  • DT 为小数类型,包括 FLOATDOUBLEDECIMALNUMERIC
  • M 为该类型数据的精度,即整数位加上小数位的总长度。
  • D 为小数位的长度。

schema 文件中对小数类型的定义如下:

{ "ColumnName":"COL1", "ColumnType":"{DT} [UNSIGNED]", "ColumnPrecision":"{M}", "ColumnScale":"{D}" }

时间和日期类型

TiDB 中的日期类型可被定义为 DT,其中:

  • DT 为日期类型,包括 DATEYEAR

schema 文件中对日期类型的定义如下:

{ "ColumnName":"COL1", "ColumnType":"{DT}" }

TiDB 中的时间类型可被定义为 TT[(M)],其中:

  • TT 为时间类型,包括 TIMEDATETIMETIMESTAMP
  • M 为秒的精度,取值范围为 0~6。

schema 文件中对时间类型的定义如下:

{ "ColumnName":"COL1", "ColumnType":"{TT}", "ColumnScale":"{M}" }

字符串类型

TiDB 中的字符串类型可被定义为 ST[(M)],其中:

  • ST 为字符串类型,包括 CHARVARCHARTEXTBINARYBLOBJSON 等。
  • M 表示字符串的最大长度。

schema 文件中对字符串类型的定义如下:

{ "ColumnName":"COL1", "ColumnType":"{ST}", "ColumnLength":"{M}" }

Enum/Set 类型

schema 文件中对 Enum/Set 类型的定义如下:

{ "ColumnName":"COL1", "ColumnType":"{ENUM/SET}", }