同步数据到存储服务
从 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-07-04T18: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":"v8.2.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/A | canal-json 和 csv |
enable-tidb-extension | protocol 参数为 canal-json 时,如果该值为 true ,TiCDC 会发送 WATERMARK 事件,并在 canal-json 消息中添加 TiDB 扩展字段。 | false | false 和 true |
配置外部存储
将数据存储到云服务存储系统时,根据云服务供应商的不同,需要设置不同的鉴权参数。本节介绍使用 Amazon S3、Google Cloud Storage (GCS) 及 Azure Blob Storage 时所用存储服务的鉴权方式以及如何配置访问相应存储服务的账户。
- Amazon S3
- GCS
- Azure Blob Storage
Amazon S3 配置样例如下:
--sink-uri="s3://bucket/prefix?protocol=canal-json"
在同步数据之前,需要为 Amazon S3 中的目录设置相应的访问权限:
- TiCDC 需要的最小权限是:
s3:ListBucket
、s3:PutObject
和s3:GetObject
。 - 如果 changefeed 的参数
sink.cloud-storage-config.flush-concurrency
大于 1,表示开启了单文件的并行上传,需要额外增加 ListParts 相关权限:s3:AbortMultipartUpload
s3:ListMultipartUploadParts
s3:ListBucketMultipartUploads
如果你还没有创建同步数据保存目录,可以参考创建存储桶在指定的区域中创建一个 S3 存储桶。如果需要使用文件夹,可以参考使用文件夹在 Amazon S3 控制台中组织对象在存储桶中创建一个文件夹。
可以通过以下两种方式配置访问 Amazon S3 的账户:
方式一:指定访问密钥
如果指定访问密钥和秘密访问密钥,将按照指定的访问密钥和秘密访问密钥进行鉴权。除了在 URI 中指定密钥外,还支持以下方式:
- 读取
$AWS_ACCESS_KEY_ID
和$AWS_SECRET_ACCESS_KEY
环境变量 - 读取
$AWS_ACCESS_KEY
和$AWS_SECRET_KEY
环境变量 - 读取共享凭证文件,路径由
$AWS_SHARED_CREDENTIALS_FILE
环境变量指定 - 读取共享凭证文件,路径为
~/.aws/credentials
- 读取
方式二:基于 IAM Role 进行访问
为运行 TiCDC Server 的 EC2 实例关联一个配置了访问 S3 访问权限的 IAM role。设置成功后,TiCDC 可以直接访问对应的 S3 中的备份目录,而不需要额外的设置。
GCS 配置样例如下:
--sink-uri="gcs://bucket/prefix?protocol=canal-json"
配置访问 GCS 的账户可以通过指定访问密钥的方式。如果指定了 credentials-file
参数,将按照指定的 credentials-file
进行鉴权。除了在 URI 中指定密钥文件外,还支持以下方式:
- 读取位于
$GOOGLE_APPLICATION_CREDENTIALS
环境变量所指定路径的文件内容 - 读取位于
~/.config/gcloud/application_default_credentials.json
的文件内容 - 在 GCE 或 GAE 中运行时,从元数据服务器中获取的凭证
Azure Blob Storage 配置样例如下:
--sink-uri="azure://bucket/prefix?protocol=canal-json"
可以通过以下方式配置访问 Azure Blob Storage 的账户:
方式一:指定共享访问签名
在 URI 中配置
account-name
和sas-token
,则使用该参数指定的存储账户名和共享访问签名令牌。由于共享访问签名令牌中带有&
的字符,需要将其编码为%26
后再添加到 URI 中。你也可以直接对整个sas-token
进行一次百分号编码。方式二:指定访问密钥
在 URI 中配置
account-name
和account-key
,则使用该参数指定的存储账户名和密钥。除了在 URI 中指定密钥文件外,还支持读取$AZURE_STORAGE_KEY
的方式。方式三:使用 Azure AD 备份恢复
运行环境配置
$AZURE_CLIENT_ID
、$AZURE_TENANT_ID
和$AZURE_CLIENT_SECRET
。
配置 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
:存储服务类型。例如:s3
、gcs
、azure
、file
。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.csv
到 CDC000004.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
:该数组表示表中每一列的详细信息。TableColumnsTotal
:TableColumns
数组的大小。
库级 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
为整数类型,包括TINYINT
、SMALLINT
、MEDIUMINT
、INT
、BIGINT
和BIT
。M
为该类型的显示宽度。
schema 文件中对整数类型定义如下:
{
"ColumnName":"COL1",
"ColumnType":"{IT} [UNSIGNED]",
"ColumnPrecision":"{M}"
}
小数类型
TiDB 中的小数类型可被定义为 DT[(M,D)][UNSIGNED]
,其中:
DT
为小数类型,包括FLOAT
、DOUBLE
、DECIMAL
和NUMERIC
。M
为该类型数据的精度,即整数位加上小数位的总长度。D
为小数位的长度。
schema 文件中对小数类型的定义如下:
{
"ColumnName":"COL1",
"ColumnType":"{DT} [UNSIGNED]",
"ColumnPrecision":"{M}",
"ColumnScale":"{D}"
}
时间和日期类型
TiDB 中的日期类型可被定义为 DT
,其中:
DT
为日期类型,包括DATE
和YEAR
。
schema 文件中对日期类型的定义如下:
{
"ColumnName":"COL1",
"ColumnType":"{DT}"
}
TiDB 中的时间类型可被定义为 TT[(M)]
,其中:
TT
为时间类型,包括TIME
、DATETIME
和TIMESTAMP
。M
为秒的精度,取值范围为 0~6。
schema 文件中对时间类型的定义如下:
{
"ColumnName":"COL1",
"ColumnType":"{TT}",
"ColumnScale":"{M}"
}
字符串类型
TiDB 中的字符串类型可被定义为 ST[(M)]
,其中:
ST
为字符串类型,包括CHAR
、VARCHAR
、TEXT
、BINARY
、BLOB
、JSON
等。M
表示字符串的最大长度。
schema 文件中对字符串类型的定义如下:
{
"ColumnName":"COL1",
"ColumnType":"{ST}",
"ColumnLength":"{M}"
}
Enum/Set 类型
schema 文件中对 Enum/Set 类型的定义如下:
{
"ColumnName":"COL1",
"ColumnType":"{ENUM/SET}",
}