- 文档中心
- 关于 TiDB
- 快速上手
- 部署标准集群
- 数据迁移
- 运维操作
- 监控与告警
- 故障诊断
- 性能调优
- 系统调优
- 软件调优
- SQL 性能调优
- SQL 性能调优概览
- 理解 TiDB 执行计划
- SQL 优化流程
- 控制执行计划
- 教程
- 同城多中心部署
- 两地三中心部署
- 同城两中心部署
- 读取历史数据
- 使用 Stale Read 功能读取历史数据(推荐)
- 使用系统变量
tidb_snapshot
读取历史数据
- 最佳实践
- Placement Rules 使用文档
- Load Base Split 使用文档
- Store Limit 使用文档
- TiDB 工具
- 功能概览
- 适用场景
- 工具下载
- TiUP
- 文档地图
- 概览
- 术语及核心概念
- TiUP 组件管理
- FAQ
- 故障排查
- TiUP 命令参考手册
- 命令概览
- TiUP 命令
- TiUP Cluster 命令
- TiUP Cluster 命令概览
- tiup cluster audit
- tiup cluster check
- tiup cluster clean
- tiup cluster deploy
- tiup cluster destroy
- tiup cluster disable
- tiup cluster display
- tiup cluster edit-config
- tiup cluster enable
- tiup cluster help
- tiup cluster import
- tiup cluster list
- tiup cluster patch
- tiup cluster prune
- tiup cluster reload
- tiup cluster rename
- tiup cluster replay
- tiup cluster restart
- tiup cluster scale-in
- tiup cluster scale-out
- tiup cluster start
- tiup cluster stop
- tiup cluster template
- tiup cluster upgrade
- TiUP DM 命令
- TiUP DM 命令概览
- tiup dm audit
- tiup dm deploy
- tiup dm destroy
- tiup dm disable
- tiup dm display
- tiup dm edit-config
- tiup dm enable
- tiup dm help
- tiup dm import
- tiup dm list
- tiup dm patch
- tiup dm prune
- tiup dm reload
- tiup dm replay
- tiup dm restart
- tiup dm scale-in
- tiup dm scale-out
- tiup dm start
- tiup dm stop
- tiup dm template
- tiup dm upgrade
- TiDB 集群拓扑文件配置
- DM 集群拓扑文件配置
- TiUP 镜像参考指南
- TiUP 组件文档
- TiDB Operator
- Dumpling
- TiDB Lightning
- TiDB Data Migration
- Backup & Restore (BR)
- TiDB Binlog
- TiCDC
- TiUniManager
- sync-diff-inspector
- TiSpark
- 参考指南
- 架构
- 监控指标
- 安全加固
- 权限
- SQL
- SQL 语言结构和语法
- SQL 语句
ADD COLUMN
ADD INDEX
ADMIN
ADMIN CANCEL DDL
ADMIN CHECKSUM TABLE
ADMIN CHECK [TABLE|INDEX]
ADMIN SHOW DDL [JOBS|QUERIES]
ADMIN SHOW TELEMETRY
ALTER DATABASE
ALTER INDEX
ALTER INSTANCE
ALTER PLACEMENT POLICY
ALTER TABLE
ALTER USER
ANALYZE TABLE
BACKUP
BEGIN
CHANGE COLUMN
CHANGE DRAINER
CHANGE PUMP
COMMIT
CREATE [GLOBAL|SESSION] BINDING
CREATE DATABASE
CREATE INDEX
CREATE PLACEMENT POLICY
CREATE ROLE
CREATE SEQUENCE
CREATE TABLE LIKE
CREATE TABLE
CREATE USER
CREATE VIEW
DEALLOCATE
DELETE
DESC
DESCRIBE
DO
DROP [GLOBAL|SESSION] BINDING
DROP COLUMN
DROP DATABASE
DROP INDEX
DROP PLACEMENT POLICY
DROP ROLE
DROP SEQUENCE
DROP STATS
DROP TABLE
DROP USER
DROP VIEW
EXECUTE
EXPLAIN ANALYZE
EXPLAIN
FLASHBACK TABLE
FLUSH PRIVILEGES
FLUSH STATUS
FLUSH TABLES
GRANT <privileges>
GRANT <role>
INSERT
KILL [TIDB]
LOAD DATA
LOAD STATS
MODIFY COLUMN
PREPARE
RECOVER TABLE
RENAME INDEX
RENAME TABLE
RENAME USER
REPLACE
RESTORE
REVOKE <privileges>
REVOKE <role>
ROLLBACK
SELECT
SET DEFAULT ROLE
SET [NAMES|CHARACTER SET]
SET PASSWORD
SET ROLE
SET TRANSACTION
SET [GLOBAL|SESSION] <variable>
SHOW [BACKUPS|RESTORES]
SHOW ANALYZE STATUS
SHOW [GLOBAL|SESSION] BINDINGS
SHOW BUILTINS
SHOW CHARACTER SET
SHOW COLLATION
SHOW [FULL] COLUMNS FROM
SHOW CONFIG
SHOW CREATE PLACEMENT POLICY
SHOW CREATE SEQUENCE
SHOW CREATE TABLE
SHOW CREATE USER
SHOW DATABASES
SHOW DRAINER STATUS
SHOW ENGINES
SHOW ERRORS
SHOW [FULL] FIELDS FROM
SHOW GRANTS
SHOW INDEX [FROM|IN]
SHOW INDEXES [FROM|IN]
SHOW KEYS [FROM|IN]
SHOW MASTER STATUS
SHOW PLACEMENT
SHOW PLACEMENT FOR
SHOW PLACEMENT LABELS
SHOW PLUGINS
SHOW PRIVILEGES
SHOW [FULL] PROCESSSLIST
SHOW PROFILES
SHOW PUMP STATUS
SHOW SCHEMAS
SHOW STATS_HEALTHY
SHOW STATS_HISTOGRAMS
SHOW STATS_META
SHOW STATUS
SHOW TABLE NEXT_ROW_ID
SHOW TABLE REGIONS
SHOW TABLE STATUS
SHOW [FULL] TABLES
SHOW [GLOBAL|SESSION] VARIABLES
SHOW WARNINGS
SHUTDOWN
SPLIT REGION
START TRANSACTION
TABLE
TRACE
TRUNCATE
UPDATE
USE
WITH
- 数据类型
- 函数与操作符
- 聚簇索引
- 约束
- 生成列
- SQL 模式
- 表属性
- 事务
- 垃圾回收 (GC)
- 视图
- 分区表
- 临时表
- 字符集和排序
- Placement Rules in SQL
- 系统表
mysql
- INFORMATION_SCHEMA
- Overview
ANALYZE_STATUS
CLIENT_ERRORS_SUMMARY_BY_HOST
CLIENT_ERRORS_SUMMARY_BY_USER
CLIENT_ERRORS_SUMMARY_GLOBAL
CHARACTER_SETS
CLUSTER_CONFIG
CLUSTER_HARDWARE
CLUSTER_INFO
CLUSTER_LOAD
CLUSTER_LOG
CLUSTER_SYSTEMINFO
COLLATIONS
COLLATION_CHARACTER_SET_APPLICABILITY
COLUMNS
DATA_LOCK_WAITS
DDL_JOBS
DEADLOCKS
ENGINES
INSPECTION_RESULT
INSPECTION_RULES
INSPECTION_SUMMARY
KEY_COLUMN_USAGE
METRICS_SUMMARY
METRICS_TABLES
PARTITIONS
PLACEMENT_RULES
PROCESSLIST
REFERENTIAL_CONSTRAINTS
SCHEMATA
SEQUENCES
SESSION_VARIABLES
SLOW_QUERY
STATISTICS
TABLES
TABLE_CONSTRAINTS
TABLE_STORAGE_STATS
TIDB_HOT_REGIONS
TIDB_HOT_REGIONS_HISTORY
TIDB_INDEXES
TIDB_SERVERS_INFO
TIDB_TRX
TIFLASH_REPLICA
TIKV_REGION_PEERS
TIKV_REGION_STATUS
TIKV_STORE_STATUS
USER_PRIVILEGES
VIEWS
METRICS_SCHEMA
- UI
- CLI
- 命令行参数
- 配置文件参数
- 系统变量
- 存储引擎
- 遥测
- 错误码
- 通过拓扑 label 进行副本调度
- 常见问题解答 (FAQ)
- 版本发布历史
- 术语表
Binlog Consumer Client 用户文档
目前 Drainer 提供了多种输出方式,包括 MySQL、TiDB、file 等。但是用户往往有一些自定义的需求,比如输出到 Elasticsearch、Hive 等,这些需求 Drainer 现在还没有实现,因此 Drainer 增加了输出到 Kafka 的功能,将 binlog 数据解析后按一定的格式再输出到 Kafka 中,用户编写代码从 Kafka 中读出数据再进行处理。
配置 Kafka Drainer
修改 Drainer 的配置文件,设置输出为 Kafka,相关配置如下:
[syncer]
db-type = "kafka"
[syncer.to]
# Kafka 地址
kafka-addrs = "127.0.0.1:9092"
# Kafka 版本号
kafka-version = "2.4.0"
自定义开发
数据格式
首先需要了解 Drainer 写入到 Kafka 中的数据格式:
// Column 保存列的数据,针对数据的类型,保存在对应的变量中
message Column {
// 数据是否为 null
optional bool is_null = 1 [ default = false ];
// 保存 int 类型的数据
optional int64 int64_value = 2;
// 保存 uint、enum, set 类型的数据
optional uint64 uint64_value = 3;
// 保存 float、double 类型的数据
optional double double_value = 4;
// 保存 bit、blob、binary、json 类型的数据
optional bytes bytes_value = 5;
// 保存 date、time、decimal、text、char 类型的数据
optional string string_value = 6;
}
// ColumnInfo 保存列的信息,包括列名、类型、是否为主键
message ColumnInfo {
optional string name = 1 [ (gogoproto.nullable) = false ];
// MySQL 中小写的列字段类型
// https://dev.mysql.com/doc/refman/8.0/en/data-types.html
// numeric 类型:int bigint smallint tinyint float double decimal bit
// string 类型:text longtext mediumtext char tinytext varchar
// blob longblob mediumblob binary tinyblob varbinary
// enum set
// json 类型:json
optional string mysql_type = 2 [ (gogoproto.nullable) = false ];
optional bool is_primary_key = 3 [ (gogoproto.nullable) = false ];
}
// Row 保存一行的具体数据
message Row { repeated Column columns = 1; }
// MutationType 表示 DML 的类型
enum MutationType {
Insert = 0;
Update = 1;
Delete = 2;
}
// Table 包含一个表的数据变更
message Table {
optional string schema_name = 1;
optional string table_name = 2;
repeated ColumnInfo column_info = 3;
repeated TableMutation mutations = 4;
}
// TableMutation 保存一行数据的变更
message TableMutation {
required MutationType type = 1;
// 修改后的数据
required Row row = 2;
// 修改前的数据,只对 Update MutationType 有效
optional Row change_row = 3;
}
// DMLData 保存一个事务所有的 DML 造成的数据变更
message DMLData {
// `tables` 包含事务中所有表的数据变更
repeated Table tables = 1;
}
// DDLData 保存 DDL 的信息
message DDLData {
// 当前使用的数据库
optional string schema_name = 1;
// 相关表
optional string table_name = 2;
// `ddl_query` 是原始的 DDL 语句 query
optional bytes ddl_query = 3;
}
// BinlogType 为 Binlog 的类型,分为 DML 和 DDL
enum BinlogType {
DML = 0; // Has `dml_data`
DDL = 1; // Has `ddl_query`
}
// Binlog 保存一个事务所有的变更,Kafka 中保存的数据为该结构数据序列化后的结果
message Binlog {
optional BinlogType type = 1 [ (gogoproto.nullable) = false ];
optional int64 commit_ts = 2 [ (gogoproto.nullable) = false ];
optional DMLData dml_data = 3;
optional DDLData ddl_data = 4;
}
查看数据格式的具体定义,参见 secondary_binlog.proto
。
Driver
TiDB-Tools 项目提供了用于读取 Kafka 中 binlog 数据的 Driver,具有如下功能:
- 读取 Kafka 的数据
- 根据 commit ts 查找 binlog 在 kafka 中的储存位置
使用该 Driver 时,用户需要配置如下信息:
- KafkaAddr:Kafka 集群的地址
- CommitTS:从哪个 commit ts 开始读取 binlog
- Offset:从 Kafka 哪个 offset 开始读取,如果设置了 CommitTS 就不用配置该参数
- ClusterID:TiDB 集群的 cluster ID
- Topic: Kafka Topic 名称,如果 Topic 名称为空,将会使用 drainer
<ClusterID>_obinlog
中的默认名称
用户以包的形式引用 Driver 的代码即可使用,可以参考 Driver 中提供的示例代码来学习如何使用 Driver 以及 binlog 数据的解析,目前提供了两个例子:
- 使用该 Driver 将数据同步到 MySQL,该示例包含将 binlog 转化为 SQL 的具体方法
- 使用该 Driver 将数据打印出来
Driver 项目地址:Binlog Slave Driver。
注意
- 示例代码仅仅用于示范如何使用 Driver,如果需要用于生产环境需要优化代码。
- 目前仅提供了 golang 版本的 Driver 以及示例代码。如果需要使用其他语言,用户需要根据 binlog 的 proto 文件生成相应语言的代码文件,并自行开发程序读取 Kafka 中的 binlog 数据、解析数据、输出到下游。也欢迎用户优化 example 代码,以及提交其他语言的示例代码到 TiDB-Tools。
文档内容是否有帮助?