Storage sink 消费程序设计

Storage sink 消费程序设计

本文介绍如何设计和实现一个 TiDB 数据变更的消费程序。

注意

当前 Storage sink 无法处理 DROP DATABASE DDL，请你尽量避免执行该语句。如果需要执行 DROP DATABASE DDL，请在下游 MySQL 手动执行。

TiCDC 不提供消费存储服务的数据的标准实现。本文介绍一个基于 Golang 的消费示例程序，该示例程序能够读取存储服务中的数据并写入到兼容 MySQL 的下游数据库。你可以参考本文提供的数据格式和以下示例代码实现消费端。

Golang 示例代码

Consumer 设计

下图是 Consumer 的整体消费流程：

以下是 Consumer 消费流程中的组件和功能定义，及其功能注释：

1. type StorageReader struct {
2. }

4. // Read the files from storage
5. // Add newly added files and delete files that not exist in storage
6. func (c *StorageReader) ReadFiles() {}

8. // Query newly added files and the latest checkpoint from storage，one file can only be returned once
9. func (c *StorageReader) ExposeNewFiles() (int64, []string) {}

12. // ConsumerManager is responsible for assigning tasks to TableConsumer.
13. // Different consumers can consume data concurrently,
14. // but data of one table must be processed by the same TableConsumer.
15. type ConsumerManager struct {
16. // StorageCheckpoint is recorded in metadata file and it can be fetched by calling `StorageReader.ExposeNewFiles()`.
17. // It indicates that the data whose transaction commit time is less than this checkpoint has been stored in storage
18. StorageCheckpoint int64
19. // it indicates where the consumer has consumed
20. // ConsumerManager periodically collects TableConsumer.Checkpoint,
21. // then Checkpoint is updated to the minimum value of all TableConsumer.Checkpoint
22. Checkpoint int64

24. tableFiles[schema][table]*TableConsumer
25. }

27. // Query newly files from StorageReader
28. // For new created table, create a TableConsumer for the new table
29. // If any, send new files to the corresponding TableConsumer
30. func (c *ConsumerManager) Dispatch() {}

33. type TableConsumer struct {
34. // it indicates where this TableConsumer has consumed
35. // Its initial value is ConsumerManager.Checkpoint
36. // TableConsumer.Checkpoint is equal to TableVersionConsumer.Checkpoint
37. Checkpoint int64

39. schema,table string
40. // Must be consumed sequentially according to the table version order
41. verConsumers map[version int64]*TableVersionConsumer
42. currentVer, previousVer int64
43. }

45. // Send newly files to the corresponding TableVersionConsumer
46. // For any DDL, assign a TableVersionConsumer for the new table version
47. func (tc *TableConsumer) Dispatch() {}

49. // If DDL query is empty or its tableVersion is less than TableConsumer.Checkpoint,
50. // - ignore this DDL, and consume the data under the table version
51. // Otherwise,
52. // - execute DDL first, and then consume the data under the table version
53. // - But for dropped table, self recycling after drop table DDL is executed
54. func (tc *TableConsumer) ExecuteDDL() {}

57. type TableVersionConsumer struct {
58. // it indicates where the TableVersionConsumer has consumed
59. // Its initial value is TableConsumer.Checkpoint
60. Checkpoint int64

62. schema,table,version string
63. // For same table version, data in different partitions can be consumed concurrently
64. # partitionNum int64
65. // Must be consumed sequentially according to the data file number
66. fileSet map[filename string]*TableVersionConsumer
67. currentVersion
68. }

70. // If data commit ts is less than TableConsumer.Checkpoint
71. // or bigger than ConsumerManager.StorageCheckpoint,
72. // - ignore this data
73. // Otherwise,
74. // - process this data and write it to MySQL
75. func (tc *TableVersionConsumer) ExecuteDML() {}

DDL 事件的处理

1. ├── metadata
2. └── test
3. ├── tbl_1
4. │   └── 437752935075545091
5. │       ├── CDC000001.json
6. │       └── schema.json

此时，首先需要解析 schema.json 文件中的表结构信息，从中获取 DDL Query 语句，并将其分为两种情况处理：

如果 DDL Query 为空或者 TableVersion 小于 consumer checkpoint 则跳过该语句。
否则，在下游 MySQL 数据库中执行获取到的 DDL 语句。

接着再开始同步数据文件 CDC000001.json。

例如，以下的 test/tbl_1/437752935075545091/schema.json 文件中 DDL Query 不为空：

1. {
2. "Table":"test",
3. "Schema":"tbl_1",
4. "Version": 1,
5. "TableVersion":437752935075545091,
6. "Query": "create table tbl_1 (Id int primary key, LastName char(20), FirstName varchar(30), HireDate datetime, OfficeLocation Blob(20))",
7. "TableColumns":[
8. {
9. "ColumnName":"Id",
10. "ColumnType":"INT",
11. "ColumnNullable":"false",
12. "ColumnIsPk":"true"
13. },
14. {
15. "ColumnName":"LastName",
16. "ColumnType":"CHAR",
17. "ColumnLength":"20"
18. },
19. {
20. "ColumnName":"FirstName",
21. "ColumnType":"VARCHAR",
22. "ColumnLength":"30"
23. },
24. {
25. "ColumnName":"HireDate",
26. "ColumnType":"DATETIME"
27. },
28. {
29. "ColumnName":"OfficeLocation",
30. "ColumnType":"BLOB",
31. "ColumnLength":"20"
32. }
33. ],
34. "TableColumnsTotal":"5"
35. }

当程序再次遍历目录，发现该表新增了一个版本目录。程序先消费完 test/tbl_1/437752935075545091 目录下的所有文件，然后再消费新目录下的数据。

1. ├── metadata
2. └── test
3. ├── tbl_1
4. │   ├── 437752935075545091
5. │   │   ├── CDC000001.json
6. │   │   └── schema.json
7. │   └── 437752935075546092
8. │   │   └── CDC000001.json
9. │   │   └── schema.json

消费逻辑跟上述一致，先解析 schema.json 文件中的表结构信息，从中获取 DDL Query 语句并按不同情况处理，然后同步数据文件 CDC000001.json。

DML 事件的处理

在处理完 DDL 事件后，可以在 {schema}/{table}/{table-version-separator}/ 目录下，根据具体的文件格式（CSV 或 Canal-JSON）并按照文件序号依次处理 DML 事件。

因为 TiCDC 提供 At Least Once 语义，可能出现重复发送数据的情况，所以需要在消费程序中比较数据事件的 commit ts 和 consumer checkpoint，并在 commit ts 小于 consumer checkpoint 的情况下进行去重处理。

Storage sink 消费程序编写指引

Storage sink 消费程序设计

Consumer 设计

DDL 事件的处理

DML 事件的处理