TiCDC Changefeed 命令行参数和配置参数

TiCDC Changefeed 命令行参数

本章节将以创建同步任务为例，介绍 TiCDC Changefeed 的命令行参数：

1. cdc cli changefeed create --server=http://10.0.10.25:8300 --sink-uri="mysql://root:123456@127.0.0.1:3306/" --changefeed-id="simple-replication-task"

1. Create changefeed successfully!
2. ID: simple-replication-task
3. Info: {"upstream_id":7178706266519722477,"namespace":"default","id":"simple-replication-task","sink_uri":"mysql://root:xxxxx@127.0.0.1:4000/?time-zone=","create_time":"2023-03-10T15:05:46.679218+08:00","start_ts":438156275634929669,"engine":"unified","config":{"case_sensitive":true,"enable_old_value":true,"force_replicate":false,"ignore_ineligible_table":false,"check_gc_safe_point":true,"enable_sync_point":true,"bdr_mode":false,"sync_point_interval":30000000000,"sync_point_retention":3600000000000,"filter":{"rules":["test.*"],"event_filters":null},"mounter":{"worker_num":16},"sink":{"protocol":"","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":"v6.5.2"}
  • --changefeed-id：同步任务的 ID，格式需要符合正则表达式 ^[a-zA-Z0-9]+(\-[a-zA-Z0-9]+)*$。如果不指定该 ID，TiCDC 会自动生成一个 UUID（version 4 格式）作为 ID。
  • --sink-uri：同步任务下游的地址，需要按照以下格式进行配置，目前 scheme 支持 mysql、tidb 和 kafka。

```

1. [scheme]://[userinfo@][host]:[port][/path]?[query_parameters]
2. ```

4. URI 中包含特殊字符时，如 `! * ' ( ) ; : @ & = + $ , / ? % # [ ]`，需要对 URI 特殊字符进行转义处理。你可以在 [URI Encoder](https://meyerweb.com/eric/tools/dencoder/) 中对 URI 进行转义。
  • --start-ts：指定 changefeed 的开始 TSO。TiCDC 集群将从这个 TSO 开始拉取数据。默认为当前时间。
  • --target-ts：指定 changefeed 的目标 TSO。TiCDC 集群拉取数据直到这个 TSO 停止。默认为空，即 TiCDC 不会自动停止。
  • --config：指定 changefeed 配置文件。

TiCDC Changefeed 配置文件说明

本章节详细介绍了同步任务的配置。

1. # 指定配置文件中涉及的库名、表名是否为大小写敏感
2. # 该配置会同时影响 filter 和 sink 相关配置，默认为 true
3. case-sensitive = true

5. # 是否输出 old value，从 v4.0.5 开始支持，从 v5.0 开始默认为 true
6. enable-old-value = true

8. # 是否开启 Syncpoint 功能，从 v6.3.0 开始支持，该功能默认关闭。
9. # 从 v6.4.0 开始，使用 Syncpoint 功能需要同步任务拥有下游集群的 SYSTEM_VARIABLES_ADMIN 或者 SUPER 权限
10. # enable-sync-point = false

12. # Syncpoint 功能对齐上下游 snapshot 的时间间隔
13. # 配置格式为 h m s，例如 "1h30m30s"
14. # 默认值为 10m，最小值为 30s
15. # sync-point-interval = "5m"

17. # Syncpoint 功能在下游表中保存的数据的时长，超过这个时间的数据会被清理
18. # 配置格式为 h m s，例如 "24h30m30s"
19. # 默认值为 24h
20. # sync-point-retention = "1h"

22. [mounter]
23. # mounter 解码 KV 数据的线程数，默认值为 16
24. # worker-num = 16

26. [filter]
27. # 忽略指定 start_ts 的事务
28. # ignore-txn-start-ts = [1, 2]

30. # 过滤器规则
31. # 过滤规则语法：https://docs.pingcap.com/zh/tidb/stable/table-filter#表库过滤语法
32. rules = ['*.*', '!test.*']

34. # 事件过滤器规则
35. # 事件过滤器的详细配置规则在下方的 Event Filter 配置规则中描述
36. # 第一个事件过滤器规则
37. [[filter.event-filters]]
38. matcher = ["test.worker"] # matcher 是一个白名单，表示该过滤规则只应用于 test 库中的 worker 表
39. ignore-event = ["insert"] # 过滤掉 insert 事件
40. ignore-sql = ["^drop", "add column"] # 过滤掉以 "drop" 开头或者包含 "add column" 的 DDL
41. ignore-delete-value-expr = "name = 'john'" # 过滤掉包含 name = 'john' 条件的 delete DML
42. ignore-insert-value-expr = "id >= 100" # 过滤掉包含 id >= 100 条件的 insert DML
43. ignore-update-old-value-expr = "age < 18" # 过滤掉旧值 age < 18 的 update DML
44. ignore-update-new-value-expr = "gender = 'male'" # 过滤掉新值 gender = 'male' 的 update DML

46. # 第二个事件过滤器规则
47. [[filter.event-filters]]
48. matcher = ["test.fruit"] # 该事件过滤器只应用于 test.fruit 表
49. ignore-event = ["drop table"] # 忽略 drop table 事件
50. ignore-sql = ["delete"] # 忽略 delete DML
51. ignore-insert-value-expr = "price > 1000 and origin = 'no where'" # 忽略包含 price > 1000 和 origin = 'no where' 条件的 insert DML

53. [sink]
54. # 对于 MQ 类的 Sink，可以通过 dispatchers 配置 event 分发器
55. # 支持 partition 及 topic（从 v6.1 开始支持）两种 event 分发器。二者的详细说明见下一节。
56. # matcher 的匹配语法和过滤器规则语法相同，matcher 匹配规则的详细说明见下一节。
57. dispatchers = [
58. {matcher = ['test1.*', 'test2.*'], topic = "Topic 表达式 1", partition = "ts" },
59. {matcher = ['test3.*', 'test4.*'], topic = "Topic 表达式 2", partition = "index-value" },
60. {matcher = ['test1.*', 'test5.*'], topic = "Topic 表达式 3", partition = "table"},
61. {matcher = ['test6.*'], partition = "ts"}
62. ]

64. # protocol 用于指定传递到下游的协议格式
65. # 当下游类型是 Kafka 时，支持 canal-json、avro 两种协议。
66. # 当下游类型是存储服务时，目前仅支持 canal-json、csv 两种协议。
67. protocol = "canal-json"

69. # 以下三个配置项仅在同步到存储服务的 sink 中使用，在 MQ 和 MySQL 类 sink 中无需设置。
70. # 换行符，用来分隔两个数据变更事件。默认值为空，表示使用 "\r\n" 作为换行符。
71. terminator = ''
72. # 文件路径的日期分隔类型。可选类型有 `none`、`year`、`month` 和 `day`。默认值为 `none`，即不使用日期分隔。详见 <https://docs.pingcap.com/zh/tidb/dev/ticdc-sink-to-cloud-storage#数据变更记录>。
73. date-separator = 'none'
74. # 是否使用 partition 作为分隔字符串，将一张表中各个 partition 的数据分不同的目录来存储。在 v6.5.0 和 v6.5.1 中，默认值为 false。在 v6.5.2 或更高的 v6.5.x 版本中，默认值为 true。建议保持该配置项为 true 以避免下游分区表可能丢数据的问题 <https://github.com/pingcap/tiflow/issues/8724>。使用示例详见 <https://docs.pingcap.com/zh/tidb/dev/ticdc-sink-to-cloud-storage#数据变更记录>。
75. enable-partition-separator = true

77. # 从 v6.5.0 开始，TiCDC 支持以 CSV 格式将数据变更记录保存至存储服务中，在 MQ 和 MySQL 类 sink 中无需设置。
78. [sink.csv]
79. # 字段之间的分隔符。必须为 ASCII 字符，默认值为 `,`。
80. delimiter = ','
81. # 用于包裹字段的引号字符。空值代表不使用引号字符。默认值为 `"`。
82. quote = '"'
83. # CSV 中列为 NULL 时将以什么字符来表示。默认值为 `\N`。
84. null = '\N'
85. # 是否在 CSV 行中包含 commit-ts。默认值为 false。
86. include-commit-ts = false