Datasets 流式读取：大型语料如何分批筛选与复现抽检

抽检范围与接入准备

编辑团队要回答的具体问题是：如何从大型文本语料中取出符合条件的一批写作样本，并让下一位编辑能够追溯和重复这次抽检。先将结论范围写成某个固定版本、某组分片、某项筛选规则内的样本观察，不直接写成整库质量结论。

资料包中的 Hugging Face 官方博客索引收录了 Streaming datasets 相关文章，条目日期为 2025 年 10 月 27 日。但索引没有文章正文、接口签名、格式支持表或运行记录，不能据此确认当前版本的具体调用方式，也不能把标题中的效率表述套用到自己的语料。

完整加载不等于必须把所有内容同时放进内存；比较时应分别记录网络传输、磁盘、缓存和峰值内存。需要先梳理字段时，可参考数据集字段确认与条件筛选流程，再回到本题建立抽检记录。

1. 登记数据集标识、配置、数据划分和不可变版本标识。若仓库引用了可变化的外部文件，还要单独记录文件版本或校验值。
2. 保存有序分片清单，不依赖每次可能变化的文件匹配结果。写清本次只包含哪些分片、排除了哪些分片。
3. 确认正文与语言字段的实际名称，查看空值、列表或嵌套对象，不把所有内容直接强转成字符串。
4. 核对所用 Datasets 版本的加载入口、流式选项、筛选与打乱接口，以及目标文件格式和压缩方式的限制。Hugging Face IterableDataset 的具体使用约定也应在此核对。
5. 先以少量授权、脱敏样本检查读取和导出，固定单进程与输入顺序后，再讨论并行和吞吐优化。

远程接入的流程示意为：固定版本与分片 → 依据已核对接口创建流式迭代入口 → 为记录补充来源定位 → 按规则筛选 → 按已核对的打乱方法抽取有限样本 → 分批写出。这里没有给出官方 API 代码，因为现有资料无法完成接口核验。

不要同时应用库内打乱和下文的本地缓冲打乱，否则实验改变了两个环节。下文仅演示自行设计的处理原型，不声称复现 Datasets 内部算法。

本地筛选与分批导出原型

以下代码使用 Python 标准库处理本地 JSONL 分片，不负责远程下载。输入文件需由团队依法取得；示例路径不是资料包提供的产物。每行必须是包含 text 和 lang 的对象，程序会补充文件路径和原始行号。JSONL 在这里只是本原型约定的输入格式，不代表已经确认远程 streaming 数据集的格式支持。

示例规则为：lang 等于 zh，且去除首尾空白后的正文长度不少于 200 个 Python 字符。它用于演示筛选，不代表语言识别或写作质量判断。每次只缓存有限记录，最多导出 300 条，每个文件最多 100 条。

import hashlib, json, platform, random
from collections import Counter
from itertools import islice
from pathlib import Path

def file_hash(path):
    h = hashlib.sha256()
    with Path(path).open('rb') as f:
        for chunk in iter(lambda: f.read(1048576), b''):
            h.update(chunk)
    return h.hexdigest()

def records(files, counts):
    for name in files:
        with Path(name).open(encoding='utf-8') as f:
            for line_no, line in enumerate(f, 1):
                row = json.loads(line)
                if not isinstance(row, dict):
                    raise ValueError((name, line_no, 'not an object'))
                counts[name] += 1
                text = row.get('text')
                if (row.get('lang') == 'zh' and
                        isinstance(text, str) and len(text.strip()) >= 200):
                    yield {'source_file': name, 'line': line_no,
                           'source_id': row.get('id'), 'text': text,
                           'lang': row['lang']}

def buffered(rows, size, seed):
    if size < 1:
        raise ValueError('buffer size must be positive')
    rng = random.Random(seed)
    it = iter(rows)
    buf = list(islice(it, size))
    for row in it:
        pos = rng.randrange(len(buf))
        yield buf[pos]
        buf[pos] = row
    rng.shuffle(buf)
    yield from buf

def export(files, out, size=1000, seed=42, limit=300):
    files = [str(Path(p).resolve()) for p in files]
    before = {p: file_hash(p) for p in files}
    out = Path(out)
    out.mkdir(parents=True, exist_ok=False)
    read_counts, selected = Counter(), Counter()
    rows = islice(buffered(records(files, read_counts), size, seed), limit)
    batch, outputs, total = [], [], 0
    def flush():
        path = out / ('samples-%03d.jsonl' % (len(outputs) + 1))
        with path.open('x', encoding='utf-8', newline='\n') as f:
            for row in batch:
                f.write(json.dumps(row, ensure_ascii=False,
                                   sort_keys=True) + '\n')
        outputs.append({'file': path.name, 'rows': len(batch),
                        'sha256': file_hash(path)})
        batch.clear()
    for row in rows:
        selected[row['source_file']] += 1
        batch.append(row)
        total += 1
        if len(batch) == 100:
            flush()
    if batch:
        flush()
    after = {p: file_hash(p) for p in files}
    if before != after:
        raise RuntimeError('input changed; reject this run')
    manifest = {
        'status': 'complete', 'prototype': 'local-buffer-v1',
        'python': platform.python_version(), 'files': files,
        'input_sha256': before, 'buffer_size': size, 'seed': seed,
        'limit': limit, 'batch_size': 100,
        'filter': {'lang': 'zh', 'min_stripped_chars': 200},
        'order': 'filter-buffer-limit-export', 'exported': total,
        'read_by_file': {p: read_counts[p] for p in files},
        'selected_by_file': {p: selected[p] for p in files},
        'outputs': outputs
    }
    (out / 'manifest.json').write_text(
        json.dumps(manifest, ensure_ascii=False, sort_keys=True, indent=2),
        encoding='utf-8')

files = ['data/part-a.jsonl', 'data/part-b.jsonl']
export(files, 'review-a', size=1000, seed=42)

该原型先筛选再缓冲，因此缓冲区装的是符合规则的记录。它在导出到上限时停止，不是全量扫描。输入哈希检查会额外顺序读取本地文件，不能把这部分开销算作远程流式接入的性能表现。

如某行是英文、正文为 null 或长度不足，就不会进入样本；缺少 lang 也会被排除。对于缺失字段，应先确认这是预期规则，避免把字段映射错误误当成没有合格样本。程序没有运行记录，不预填任何导出数量或对照结果。

缓冲区与种子对照

下面是待执行的对照设计，数值只作为实验参数，不是官方建议。每次使用新的输出目录，从同一组不可变输入重新开始。

1. 分别调用 export，输出到 review-a、review-b、review-c、review-d；保持筛选规则、样本上限、文件顺序和 Python 环境不变。
2. 对 A 与 B 比较每个样本文件的哈希，并按文件顺序逐条比较 source_file 与 line。相同样本集合但顺序不同，不算有序输出一致。
3. 对 A 与 C、A 与 D 计算来源定位集合的交集和并集；并集非空时，用交集数量除以并集数量记录集合重叠率。空结果单独标记，不进行除法。
4. 比较各文件选中条数、正文长度区间和内容类别。不同设置也可能得到相同结果，不要把样本必须变化作为通过条件。

固定种子只是复现条件之一。数据版本、文件顺序、筛选代码、库版本、进程数量或恢复位置变化，都应作为新实验登记。接入库内打乱后，还需要按照其已核对的语义重做对照，不能沿用本原型结果。

分片覆盖与样本交付

manifest.json 保存输入清单、校验值、处理顺序、参数、读到的记录数与选中数；samples 文件保留正文与来源位置。只有全部执行成功才产生完成清单。中断目录即使已有样本文件，也不能按完整交付使用；本原型不提供断点续跑。

read_by_file 是迭代器实际解析的原始行数，不是网络字节量。缓冲区可能提前读取尚未导出的记录，因此读到不等于选中。selected_by_file 为零也不等于该分片没有合格内容，可能只是没有读到那里或没有入样。

对照完整分片清单，将每个分片标为未触达、已读取部分或已完整扫描。原型中的计数本身不能判断某文件是否已完整扫描，需要结合独立行数基准或补充结束事件记录。没有可信的完整分母时，只报告已观察范围，不计算全库覆盖率。

有限缓冲打乱不能自动保证全库等概率入样。尤其是提前停止时，后方未触达分片没有机会进入本次样本。若要加强分片覆盖，可另立实验，对每个分片设置审核名额；但等额分片抽检也不等于按全库记录数等概率抽样，报告中必须写明这一差别。

常见故障排查

发布前验收

• 来源说明仅确认官方索引收录相关文章，没有把未取得的正文写成已核对依据。
• 正式接入前已补充当前 Datasets 版本、流式接口、格式与压缩限制的核对记录；未补齐时，只交付本地原型与待核对清单。
• 数据版本、授权范围、有序分片清单与输入校验值齐全，筛选与处理顺序明确。
• 样本可逐条解析，正文无意外删减，来源文件及原始行号可以回查；清单计数与实际文件一致。
• 相同条件重复运行的比较已完成并留档；未执行的实验仍标为待执行，没有填写虚构结果。
• 缓冲区与种子对照分别只改变一个变量，包含分片选中情况和未触达范围说明。
• 发布用样本已完成隐私与授权复核，原始敏感正文和访问凭据没有进入公开附件。
• 结论明确限定于本次观察范围，不承诺固定磁盘节省比例，不把缓冲打乱称为全库均匀抽样。

下一步应先补充 Streaming datasets 文章正文与当前接口文档，再把远程读取接到同一套来源定位和导出契约中。只有复现与覆盖检查通过后，才将样本观察交给编辑用于质量判断。