大模型训练数据管理的数据库挑战
大模型的训练数据集规模从数亿到数百亿条记录不等。以Kimi K3为例，其训练语料超过15TB，清洗后的有效数据约10亿条文本记录。这些数据需要在数据库中完成去重、质量标注、版本管理和溯源追踪。MySQL 8.0作为主流的关系型数据库，在这个数据规模下面临的核心问题不是存储容量，而是单表查询性能、写入吞吐和DDL操作锁表时间。单表超过5亿行后，即便是索引查询，P99延迟也会从毫秒级退化到秒级。
分库分表是解决这类问题的标准方案，但大模型训练数据的特殊性在于：数据写入是批量导入而非逐条插入，查询模式以范围扫描和聚合统计为主而非精确查找，且数据有明确的版本和分片维度。本文以训练语料管理为场景，给出完整的分库分表实战方案。
数据模型与分片策略设计
训练语料的核心表结构：
CREATE TABLE training_corpus (
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
content_hash CHAR(32) NOT NULL COMMENT '内容MD5哈希，用于去重',
content MEDIUMTEXT NOT NULL COMMENT '原始文本内容',
token_count INT UNSIGNED NOT NULL COMMENT 'Token数量',
language ENUM('zh','en','ja','ko','code','mixed') NOT NULL,
quality_score TINYINT UNSIGNED NOT NULL COMMENT '质量评分0-100',
source VARCHAR(255) NOT NULL COMMENT '数据来源标识',
dataset_version VARCHAR(32) NOT NULL COMMENT '数据集版本号',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (id),
UNIQUE KEY uk_content_hash (content_hash),
KEY idx_language_quality (language, quality_score),
KEY idx_dataset_version (dataset_version),
KEY idx_source (source)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;分片维度选择：训练数据按dataset_version版本分库，按language分表。这种设计的好处是：同一版本的数据在同一库中，跨版本的对比查询走跨库路由；同语言的数据在同一表中，按语言过滤的查询不需要跨表扫描。
分库规则：dataset_version哈希 % 8 → db_0 ~ db_7
分表规则：language编码映射 → table后缀
zh → _00, en → _01, ja → _02, ko → _03,
code → _04, mixed → _05
最终数据分布：
db_0.corpus_zh_00, db_0.corpus_en_01, ...
db_1.corpus_zh_00, db_1.corpus_en_01, ...
...
db_7.corpus_zh_00, db_7.corpus_en_01, ...
共 8库 × 6表 = 48个物理表ShardingSphere分片配置
使用ShardingSphere-JDBC 5.5作为分片中间件，对应用层透明：
# application-sharding.yml
mode:
type: Standalone
repository:
type: JDBC
dataSources:
ds_0:
url: jdbc:mysql://10.0.1.10:3306/corpus_db_0?useSSL=false
username: corpus_app
password: ${DB_PASSWORD}
connectionTimeoutMilliseconds: 5000
maxPoolSize: 50
minPoolSize: 10
ds_1:
url: jdbc:mysql://10.0.1.11:3306/corpus_db_1?useSSL=false
# ... 同上配置
# ds_2 ~ ds_7 略
rules:
- !SHARDING
tables:
training_corpus:
actualDataNodes: ds_${0..7}.corpus_${(0..5).toString().padLeft(2,'0')}
databaseStrategy:
hint:
algorithmClassName: com.example.shard.VersionHintAlgorithm
tableStrategy:
standard:
shardingColumn: language
shardingAlgorithmName: language-mod
keyGenerateStrategy:
column: id
keyGeneratorName: snowflake
shardingAlgorithms:
language-mod:
type: MOD
props:
sharding-count: '6'
logic-table: training_corpus
keyGenerators:
snowflake:
type: SNOWFLAKE
props:
worker-id: '1'Hint分片算法实现——根据dataset_version动态路由到目标库：
public class VersionHintAlgorithm implements HintShardingAlgorithm<String> {
private static final int DB_COUNT = 8;
@Override
public Collection<String> doSharding(
Collection<String> availableTargetNames,
HintShardingValue<String> shardingValue) {
String version = shardingValue.getValues().iterator().next();
int dbIndex = Math.abs(version.hashCode()) % DB_COUNT;
String target = "ds_" + dbIndex;
if (availableTargetNames.contains(target)) {
return List.of(target);
}
// fallback到默认库
return List.of("ds_0");
}
@Override
public void init(Properties props) {}
@Override
public String getType() {
return "VERSION_HINT";
}
}批量导入性能优化
训练数据导入是典型的批量写场景。单条INSERT在分片环境下性能极差，需要优化：
public class BulkDataImporter {
private final DataSource dataSource;
private static final int BATCH_SIZE = 5000;
private static final int PARALLELISM = 8;
public void importCorpus(String version, List<CorpusRecord> records) {
// 按language分组
Map<String, List<CorpusRecord>> grouped = records.stream()
.collect(Collectors.groupingBy(CorpusRecord::getLanguage));
// 并行导入不同语言的数据
grouped.entrySet().parallelStream().forEach(entry -> {
String lang = entry.getKey();
List<CorpusRecord> batch = entry.getValue();
// 分批
List<List<CorpusRecord>> partitions =
Lists.partition(batch, BATCH_SIZE);
for (List<CorpusRecord> partition : partitions) {
String sql = buildInsertSql(partition, version, lang);
try (Connection conn = dataSource.getConnection();
Statement stmt = conn.createStatement()) {
stmt.execute(sql);
} catch (SQLException e) {
// 重试逻辑
retryInsert(partition, version, lang, 3);
}
}
});
}
private String buildInsertSql(List<CorpusRecord> batch,
String version, String lang) {
StringBuilder sb = new StringBuilder();
sb.append("INSERT INTO corpus_").append(lang)
.append(" (content_hash, content, token_count, ")
.append("language, quality_score, source, dataset_version) VALUES ");
for (int i = 0; i < batch.size(); i++) {
CorpusRecord r = batch.get(i);
if (i > 0) sb.append(",");
sb.append("('").append(r.getContentHash()).append("',")
.append("'").append(escapeSql(r.getContent())).append("',")
.append(r.getTokenCount()).append(",")
.append("'").append(r.getLanguage()).append("',")
.append(r.getQualityScore()).append(",")
.append("'").append(r.getSource()).append("',")
.append("'").append(version).append("')");
}
// ON DUPLICATE KEY跳过重复数据
sb.append(" ON DUPLICATE KEY UPDATE ")
.append("updated_at=VALUES(updated_at)");
return sb.toString();
}
}MySQL 8.0的批量导入还需要以下参数调优：
-- 导入专用会话参数
SET SESSION innodb_buffer_pool_size = 8589934592; -- 8GB
SET SESSION innodb_flush_log_at_trx_commit = 2;
SET SESSION sync_binlog = 0;
SET SESSION unique_checks = 0;
SET SESSION foreign_key_checks = 0;
SET SESSION tx_isolation = 'READ-COMMITTED';
-- 导入完成后恢复
SET SESSION innodb_flush_log_at_trx_commit = 1;
SET SESSION sync_binlog = 1;
SET SESSION unique_checks = 1;跨库聚合查询方案
按语言统计各版本的Token总量是常见需求，但跨8个库做聚合的性能很差。方案是维护一张汇总表，通过定时任务增量更新：
CREATE TABLE corpus_stats_daily (
stat_date DATE NOT NULL,
dataset_version VARCHAR(32) NOT NULL,
language VARCHAR(10) NOT NULL,
record_count BIGINT UNSIGNED NOT NULL DEFAULT 0,
total_tokens BIGINT UNSIGNED NOT NULL DEFAULT 0,
avg_quality_score DECIMAL(5,2) NOT NULL DEFAULT 0,
PRIMARY KEY (stat_date, dataset_version, language)
) ENGINE=InnoDB;汇总任务实现：
@Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点
public void aggregateStats() {
String yesterday = LocalDate.now().minusDays(1)
.format(DateTimeFormatter.ISO_LOCAL_DATE);
// 并行查询8个库
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (int db = 0; db < 8; db++) {
final int dbIndex = db;
futures.add(CompletableFuture.runAsync(() -> {
String sql = String.format(
"SELECT dataset_version, language, " +
"COUNT(*) as cnt, SUM(token_count) as tokens, " +
"AVG(quality_score) as avg_q " +
"FROM corpus_%%s " +
"WHERE DATE(created_at) = '%s' " +
"GROUP BY dataset_version, language",
yesterday
);
for (String lang : LANGUAGES) {
String fullSql = String.format(sql, lang);
List<Map<String, Object>> results =
jdbcTemplate.queryForList(fullSql);
upsertStats(yesterday, results);
}
}, statsExecutor));
}
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.orTimeout(30, TimeUnit.MINUTES)
.join();
}数据去重策略：布隆过滤器加速
训练语料的去重是最耗时的操作。content_hash唯一键可以防止数据库层面的重复，但需要在应用层快速判断一条数据是否已存在。布隆过滤器是标准方案：
public class DedupFilter {
// 预期10亿条，误判率0.01%
private final BloomFilter<String> filter;
private static final long EXPECTED_INSERTIONS = 1_000_000_000L;
private static final double FPP = 0.0001;
public DedupFilter() {
this.filter = BloomFilter.create(
Funnels.stringFunnel(Charsets.UTF_8),
EXPECTED_INSERTIONS,
FPP
);
}
public boolean mightExist(String contentHash) {
return filter.mightContain(contentHash);
}
public void record(String contentHash) {
filter.put(contentHash);
}
// 从数据库加载已有hash到过滤器
public void warmUp(JdbcTemplate jdbc) {
String sql = "SELECT content_hash FROM training_corpus";
jdbc.query(sql, rs -> {
filter.put(rs.getString(1));
});
}
}预热阶段需要全表扫描content_hash列。对于10亿行数据，只查询单列约需30-60分钟。布隆过滤器常驻内存约2.4GB（10亿条/0.01%误判率），用Guava的BloomFilter实现。
分库分表后的运维要点
- DDL变更：48个物理表的结构变更通过pt-online-schema-change或gh-ost执行，逐库逐表滚动变更，避免全局锁表。
- 数据备份：按库粒度做mysqldump或Percona XtraBackup，每个库独立备份窗口，错峰执行。
- 容量扩容：当单表超过20亿行时，需要增加分片数。通过双写方案实现在线扩容：新写双写新老分片，存量数据异步迁移，读请求逐步切换。
- 慢查询治理：ShardingSphere的SQL审计日志记录每条SQL的路由目标，快速定位是哪个分片拖慢了整体查询。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/mysql80-zai-da-mo-xing-xun-lian-shu-ju-guan-li-zhong-de-fen/
