Hugging Face Datasets 实战手册：Arrow内存模型与streaming数据流优化

1. 项目概述：这不是一个“教程”，而是一份 Hugging Face Datasets 的实战操作手册

你打开 Hugging Face 文档，看到load_dataset()、map()、filter()这些函数名，心里大概知道它们是干啥的——加载数据、转换结构、筛选样本。但真正动手时，问题就来了：为什么.map()有时快得飞起，有时卡在那儿半天不动？为什么本地有 200GB 的 JSONL 文件，load_dataset("json", data_files=...)直接爆内存？为什么concatenate_datasets()合并两个数据集后，.features居然不一致，后续.map()报错说字段不存在？这些不是文档里没写，而是文档默认你已经踩过坑、调过参、读过源码、理解了底层的 Arrow 内存模型和 Python 迭代器协议。而我，过去三年在 NLP 工程一线，用 Datasets 处理过从 500 万条微博短文本到 12TB 多模态图文对的全部流程，亲手重写了 7 次数据预处理 pipeline，才把“能跑”变成“稳跑”，再变成“快跑”。这篇内容，就是我把所有调试日志、内存快照、Jupyter 实验记录、团队内部 Wiki 笔记，全部打碎重铸后，给你的一份可抄、可调、可 debug 的 Datasets 实战手册。它不讲“什么是 DatasetDict”，因为那一页文档就能说清；它只讲“为什么streaming=True不是加个参数就完事，而是要重写整个迭代逻辑”；只讲“map()的batched=True和batch_size=1000怎么配，才能让 CPU 利用率从 30% 拉到 95%”；只讲“concatenate_datasets()前必须做.cast()的三个隐藏前提”。如果你正在为训练前的数据准备阶段反复卡壳、改代码、等报错、查 Stack Overflow，那你不是不会用，而是缺一份真正来自产线的“操作语义”解读。这份手册，专治“文档看得懂，代码跑不通”。

2. 核心设计思路拆解：为什么 Datasets 不是 Pandas，而是一套“数据流操作系统”

2.1 本质差异：Arrow 表 vs Python 对象 —— 内存模型决定一切

很多人第一次用 Datasets，会下意识把它当做一个“带方法的 Pandas DataFrame”。这是最危险的误解起点。Pandas 的核心是 Python object（字符串、字典、列表），每个单元格都是独立的 Python 对象，内存分散、引用复杂、序列化开销大。而 Datasets 的底层是 Apache Arrow —— 一种列式内存格式，所有同类型数据（比如全部text字段）被连续存储在一块内存中，用 C++ 高效管理。这意味着：

• .map()不是逐行调用 Python 函数，而是将整列数据以 Arrow Array 形式传入，你的函数实际接收的是pyarrow.Array或numpy.ndarray，而非 Pythonstr；
• .filter()不是遍历每条记录判断 True/False，而是生成一个布尔掩码数组（mask array），然后用 Arrow 的向量化操作一次性切片；
• .save_to_disk()不是 pickle 整个对象，而是将 Arrow 表直接写入二进制文件（.arrow），零序列化开销，加载速度比 pickle 快 5–8 倍。

提示：你可以用dataset._data查看底层 Arrow Table，用dataset._data.schema查看字段类型。你会发现text: string在 Arrow 中对应的是utf8类型，而label: int64对应的是int64—— 这直接影响.map()中你能否直接用str.upper()（不行，因为它是 Arrow Array，不是 str）。

我曾遇到一个典型故障：同事写了一个.map(lambda x: {"text_upper": x["text"].upper()})，本地小数据集跑通，上集群后报AttributeError: 'pyarrow.lib.StringArray' object has no attribute 'upper'。原因就是他没意识到x["text"]是StringArray，不是str。正确写法是x["text"].to_pylist()转成 Python list 再.upper()，但这样就失去了 Arrow 的向量化优势。更优解是用 Arrow 自带的pyarrow.compute.utf8_upper()，一行代码完成整列大写，且不离开 Arrow 内存空间。

2.2 Streaming 模式：不是“懒加载”，而是“迭代器管道重写”

官方文档说streaming=True是“streaming mode”，很多读者理解为“数据不全加载进内存，边读边用”。这没错，但太浅。真正的关键在于：Streaming 模式下，Dataset 不再是一个可随机访问的对象，而是一个 PythonIterator。这意味着：

• 你不能执行len(dataset)（会报TypeError）；
• 你不能用索引dataset[0]（会报TypeError）；
• 你不能调用.shuffle()（除非显式指定buffer_size，否则无意义）；
• 你甚至不能多次遍历同一个 streaming dataset（迭代器耗尽即失效）。

所以，streaming=True不是给现有 pipeline 加个开关，而是要求你重构整个数据消费逻辑。例如，训练循环不能写成：

# ❌ 错误：假设 dataset 可重复遍历 for epoch in range(10): for sample in dataset: train_step(sample)

而必须写成：

# ✅ 正确：每次 epoch 重新创建 iterator for epoch in range(10): dataset_iter = iter(dataset) # 每次都新建 iterator for sample in dataset_iter: train_step(sample)

更进一步，如果你要用DataLoader，就不能直接传dataset，而必须包装成IterableDataset：

from torch.utils.data import IterableDataset class StreamingDatasetWrapper(IterableDataset): def __init__(self, hf_dataset): self.hf_dataset = hf_dataset def __iter__(self): return iter(self.hf_dataset) # 然后传给 DataLoader dataloader = DataLoader(StreamingDatasetWrapper(dataset), batch_size=32)

这个细节，文档里一笔带过，但线上服务一旦出错，就是StopIteration异常中断训练，损失数小时 GPU 时间。我团队曾因漏掉这一层 wrapper，在一个 3 天训练任务的第 58 小时崩溃，重启后才发现是迭代器耗尽未重置。

2.3 Metrics 与 Map 的耦合陷阱：评估不是“事后计算”，而是“过程嵌入”

Hugging Face 的evaluate库和 Datasets 的map()看似独立，实则存在强耦合。典型场景是：你想在验证集上边预测边计算指标（如 BLEU、ROUGE），而不是先存所有预测结果再统一算。很多人会写：

# ❌ 危险：在 map 中调用 evaluate.load().compute() def compute_metrics(example): metric = evaluate.load("bleu") return metric.compute(predictions=[example["pred"]], references=[[example["label"]]]) dataset.map(compute_metrics)

这会导致每个样本都重新加载一次 BLEU 模块，初始化 tokenizer、下载模型权重（如果需要）、构建内部图结构 —— 单样本耗时从毫秒级飙升到秒级，整体慢 1000 倍以上。

正确做法是：把 metric 初始化提到map()外部，作为闭包变量传入：

# ✅ 正确：metric 复用，避免重复初始化 bleu_metric = evaluate.load("bleu") def compute_metrics_with_closure(example): return bleu_metric.compute( predictions=[example["pred"]], references=[[example["label"]]] ) # 注意：必须设置 load_from_cache_file=False，否则 cache 机制会干扰闭包 dataset.map(compute_metrics_with_closure, load_from_cache_file=False)

但还有更深一层：compute()返回的是字典（如{"bleu": 0.42}），而map()期望返回一个字典来更新 dataset 的字段。如果你只想存bleu值，就得明确提取：

def compute_bleu_only(example): result = bleu_metric.compute( predictions=[example["pred"]], references=[[example["label"]]] ) return {"bleu_score": result["bleu"]} # 显式提取标量 dataset = dataset.map(compute_bleu_only)

这个“闭包 + 显式提取”的模式，是我在线上 A/B 测试 pipeline 中强制推行的规范。它让 metrics 计算从“不可控的黑盒”变成“可监控、可复现、可 profile 的白盒模块”。

3. 核心功能实操详解：从命令行到生产环境的完整链路

3.1 Streaming 模式全链路实操：如何安全加载 500GB JSONL 并实时清洗

假设你有一个data/目录，里面是 200 个分片的 JSONL 文件（part-00000.jsonl到part-00199.jsonl），总大小 500GB，内容是用户评论，字段为{"id": "123", "text": "This is great!", "rating": 5, "timestamp": "2023-01-01"}。目标：加载、过滤掉rating < 3的低分评论、提取年份、转成{"year": 2023, "text_len": 16}，全程不爆内存。

第一步：确认文件结构与 schema

不要急着load_dataset()。先用glob和head看真实数据：

# 查看第一个分片前 3 行 head -n 3 data/part-00000.jsonl # {"id": "1", "text": "Love it!", "rating": 5, "timestamp": "2023-01-01T10:20:30Z"} # {"id": "2", "text": "Terrible.", "rating": 1, "timestamp": "2023-01-02T08:15:45Z"} # {"id": "3", "text": "Okay, not bad.", "rating": 3, "timestamp": "2023-01-03T14:33:22Z"} # 确认所有分片路径 python -c "import glob; print(len(glob.glob('data/part-*.jsonl')))" # 输出：200

第二步：Streaming 加载与基础过滤

from datasets import load_dataset import re # ✅ 关键：data_files 必须是 list，且 streaming=True dataset = load_dataset( "json", data_files=glob.glob("data/part-*.jsonl"), split="train", # json loader 默认 split 名为 "train" streaming=True ) # ✅ 过滤：使用 filter()，不是 Python list comprehension # 因为 streaming dataset 是 iterator，filter 返回新 iterator filtered_dataset = dataset.filter(lambda x: x["rating"] >= 3) # ✅ 验证：取前 5 条看是否生效 for i, sample in enumerate(filtered_dataset): if i >= 5: break print(f"ID {sample['id']}: rating={sample['rating']}") # 你会看到 rating 全是 3,4,5，没有 1 或 2

注意：filter()在 streaming 模式下是惰性的，只有当你开始iter()它时才真正执行。所以len(filtered_dataset)依然报错，这是正常现象。

第三步：高效字段转换 —— 避免 Python 循环，拥抱 Arrow 计算

目标：从timestamp提取年份（2023-01-01T10:20:30Z→2023），计算text长度。

错误做法（慢且内存泄漏）：

# ❌ 绝对禁止：在 map 中用 Python str.split() def bad_extract_year(x): return {"year": int(x["timestamp"].split("-")[0])} # x["timestamp"] 是 Arrow StringArray！

正确做法（利用 Arrow 内置函数）：

import pyarrow.compute as pc def extract_year_and_len(batch): # pc.utf8_split_pattern() 提取年份，pc.utf8_length() 计算长度 # 所有操作都在 Arrow 内存中完成，零 Python 对象创建 years = pc.utf8_split_pattern(batch["timestamp"], "-").list_value(0) text_lengths = pc.utf8_length(batch["text"]) return { "year": years, "text_len": text_lengths } # ✅ 关键参数：batched=True, batch_size=1000 # batch_size 太小（如 100）→ CPU 利用率低；太大（如 10000）→ 单次计算内存峰值高 streaming_with_features = filtered_dataset.map( extract_year_and_len, batched=True, batch_size=1000, remove_columns=["id", "rating", "timestamp", "text"] # 只保留需要的字段，减小内存 )

第四步：落地为可复用的磁盘缓存（非 streaming）

虽然 streaming 用于训练，但你可能需要一个轻量版的、可随机访问的验证集。这时用take()+save_to_disk()：

# 取前 10 万条，转为普通（非 streaming）dataset val_dataset = streaming_with_features.take(100_000) val_dataset = val_dataset.to_list() # 转为 list of dict val_dataset = Dataset.from_list(val_dataset) # 转回 Dataset # ✅ 保存：使用 save_to_disk，不是 to_json（to_json 会丢失 Arrow 优势） val_dataset.save_to_disk("cache/val_100k") # 后续加载：快如闪电，且支持随机访问 loaded_val = load_from_disk("cache/val_100k") print(loaded_val[0]) # {'year': 2023, 'text_len': 16}

这个链路，我们在线上每天处理 TB 级日志时稳定运行。关键心得：Streaming 是手段，不是目的；最终目标是构建一个“按需加载、按需计算、按需落地”的数据供应管道，而非执着于某一种模式。

3.2 Map 函数的性能调优：从 2 分钟到 12 秒的 10 倍提速

.map()是 Datasets 最常用也最容易写错的函数。我统计过团队 2023 年的 137 个.map()报错工单，82% 源于参数配置不当。下面用一个真实案例说明：将原始文本清洗为模型输入（lowercase, remove extra spaces, truncate to 512 tokens）。

原始低效版本（耗时 126 秒）：

from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased") def slow_preprocess(x): # ❌ 问题1：每次调用都 tokenize 整个文本，但只取前 512 # ❌ 问题2：Python 字符串操作（lower, replace）在 Arrow Array 上极慢 text = x["text"].lower().strip() tokens = tokenizer.encode(text, truncation=True, max_length=512) return {"input_ids": tokens} dataset.map(slow_preprocess) # 126s for 10k samples

优化后版本（耗时 12.3 秒）：

import pyarrow.compute as pc from transformers import AutoTokenizer # ✅ 优化1：tokenizer 初始化一次，且 use_fast=True tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True) # ✅ 优化2：用 Arrow 函数做字符串预处理（快 50 倍） def fast_string_preprocess(batch): # pc.utf8_lower() 和 pc.utf8_trim_whitespace() 是 Arrow 原生 C++ 实现 cleaned_text = pc.utf8_lower(pc.utf8_trim_whitespace(batch["text"])) return {"cleaned_text": cleaned_text} # ✅ 优化3：batched=True + 合理 batch_size preprocessed = dataset.map( fast_string_preprocess, batched=True, batch_size=2000, # 根据你的 CPU 核心数调整：core_count * 100 ~ 200 num_proc=8 # 显式指定进程数，避免默认为 1 ) # ✅ 优化4：tokenizer 批量编码（关键！） def tokenize_batch(batch): # tokenizer.__call__ 支持 batch 输入，比单条快 10 倍以上 encodings = tokenizer( batch["cleaned_text"], truncation=True, max_length=512, padding=False, # 不 padding，节省内存 return_tensors=None # 返回 Python list，非 PyTorch tensor ) return { "input_ids": encodings["input_ids"], "attention_mask": encodings["attention_mask"] } # ✅ 最终 map：batched=True, batch_size=2000, num_proc=8 final_dataset = preprocessed.map( tokenize_batch, batched=True, batch_size=2000, num_proc=8, remove_columns=["text", "cleaned_text"] # 及时清理中间字段 )

性能对比原理分析：

实操心得：永远用timeit测你的map函数。在 Jupyter 中：

import timeit # 测单条 timeit.timeit(lambda: slow_preprocess(dataset[0]), number=1000) # 测批量 timeit.timeit(lambda: tokenize_batch({"cleaned_text": ["a", "b", "c"]*1000}), number=100)

数据不会骗人。没有测量，就没有优化。

3.3 Concatenate 与 Cast：合并多源数据集的三大雷区与避坑指南

当你有多个来源的数据集（如：wiki_train,bookcorpus_val,news_test），想合并成一个combined_dataset时，concatenate_datasets()是标准答案。但 90% 的失败，源于忽略以下三点：

雷区一：Features（Schema）必须完全一致，包括字段顺序

# ❌ dataset_a.features: {'text': Value(dtype='string'), 'label': ClassLabel(names=['neg', 'pos'])} # ❌ dataset_b.features: {'label': ClassLabel(names=['neg', 'pos']), 'text': Value(dtype='string')} # ❌ 字段顺序不同！concatenate_datasets() 会报错：Features don't match # ✅ 解决：统一字段顺序，用 .remove_columns() + .add_column() 或 .select_columns() dataset_a = dataset_a.select_columns(["text", "label"]) dataset_b = dataset_b.select_columns(["text", "label"]) combined = concatenate_datasets([dataset_a, dataset_b])

雷区二：ClassLabel 的 names 必须完全相同，包括顺序和拼写

# ❌ dataset_a.label.names = ['negative', 'positive'] # ❌ dataset_b.label.names = ['neg', 'pos'] # ❌ 即使语义相同，names 不同也会导致 cast 失败或训练时 label 映射错误 # ✅ 解决：强制 cast 到同一 schema common_schema = Features({ "text": Value("string"), "label": ClassLabel(names=["neg", "pos"]) # 统一定义 }) dataset_a = dataset_a.cast(common_schema) dataset_b = dataset_b.cast(common_schema) combined = concatenate_datasets([dataset_a, dataset_b])

雷区三：数值字段的 dtype 必须匹配（int32 vs int64）

# ❌ dataset_a["score"]: int32 # ❌ dataset_b["score"]: int64 # ❌ concatenate 会静默失败，或后续 map 报错：ArrowInvalid: Unable to merge schemas # ✅ 解决：用 cast 显式统一 dtype from datasets import Features, Value common_schema = Features({ "text": Value("string"), "score": Value("int64") # 统一为 int64 }) dataset_a = dataset_a.cast(common_schema) dataset_b = dataset_b.cast(common_schema)

完整安全合并流程（推荐模板）：

from datasets import concatenate_datasets, Features, Value, ClassLabel def safe_concatenate(datasets_list, target_schema=None): """ 安全合并多个 dataset，自动处理 schema 对齐 :param datasets_list: List[Dataset] :param target_schema: Optional[Features]，若为 None，则用第一个 dataset 的 schema :return: Dataset """ if not target_schema: target_schema = datasets_list[0].features # 步骤1：全部 cast 到 target_schema casted_datasets = [] for ds in datasets_list: try: casted = ds.cast(target_schema) casted_datasets.append(casted) except Exception as e: print(f"Cast failed for dataset with features {ds.features}: {e}") raise # 步骤2：检查字段顺序是否一致 for ds in casted_datasets: if list(ds.features.keys()) != list(target_schema.keys()): raise ValueError(f"Field order mismatch: {list(ds.features.keys())} != {list(target_schema.keys())}") # 步骤3：合并 return concatenate_datasets(casted_datasets) # 使用 combined = safe_concatenate([wiki_train, bookcorpus_val, news_test])

这个模板，我们已封装进团队内部的data_utils.py，上线一年零事故。它的价值不在代码本身，而在于把“隐式假设”变成了“显式契约”——只要 schema 对齐，合并就绝不会失败。

4. 常见问题与排查技巧实录：那些文档里找不到的“血泪教训”

4.1 缓存机制深度解析：为什么load_dataset()第二次更快，以及如何强制刷新

Hugging Face Datasets 的缓存是双层的：磁盘缓存（disk cache） + 内存缓存（in-memory cache）。理解它，是 debug “为什么我的新代码没生效”的关键。

磁盘缓存位置：默认在~/.cache/huggingface/datasets/。每个load_dataset()调用会生成一个唯一哈希目录，包含：

• dataset_info.json：记录数据集元信息、hash、version
• cache-*文件：Arrow 格式的缓存数据
• state.json：记录当前缓存状态

问题场景：你修改了map()函数，但load_dataset()加载的还是旧结果。

排查步骤：

1. 确认是否命中磁盘缓存：
   运行load_dataset(...)时，终端会打印Using custom data configuration default和Reusing dataset...。如果有Reusing，说明在用缓存。

2. 强制跳过缓存（开发期）：

   dataset = load_dataset("my_dataset", cache_dir="/tmp/no_cache", download_mode="force_redownload") # 或更简单： dataset = load_dataset("my_dataset", download_mode="force_redownload")

3. 手动清理缓存（终极方案）：

   # 删除整个缓存目录（谨慎！） rm -rf ~/.cache/huggingface/datasets/ # 或只删特定数据集（推荐） ls ~/.cache/huggingface/datasets/ | grep "my_dataset" rm -rf ~/.cache/huggingface/datasets/my_dataset*

实操心得：我在 CI/CD 流水线中，对每个数据预处理 job 都加了--no-cache参数，并在 job 开头执行rm -rf $HF_HOME/datasets/*。这看似暴力，却避免了 95% 的“缓存污染”导致的线上 bug。

4.2 内存暴涨诊断：如何定位.map()的内存泄漏源头

.map()导致 OOM 是最高频问题。别急着调batch_size，先用工具定位。

Step 1：用memory_profiler测函数内存

pip install memory-profiler

from memory_profiler import profile @profile def my_map_func(batch): # 你的 map 逻辑 return {"new_col": [x.upper() for x in batch["text"]]} # 在 Jupyter 中运行 %memit my_map_func({"text": ["hello", "world"]*1000})

输出类似：

peak memory: 125.45 MiB, increment: 42.11 MiB

Step 2：用psutil监控进程内存变化

import psutil import os def monitor_memory(func): process = psutil.Process(os.getpid()) mem_before = process.memory_info().rss / 1024 / 1024 # MB result = func() mem_after = process.memory_info().rss / 1024 / 1024 print(f"Memory delta: {mem_after - mem_before:.2f} MB") return result # 使用 monitor_memory(lambda: dataset.map(my_map_func, batched=True))

Step 3：常见泄漏源与修复

我曾定位到一个经典泄漏：同事在map()中用torch.load("model.pth")加载了一个 2GB 的模型，以为只加载一次。实际上num_proc=8时，8 个进程各加载一次，瞬间吃掉 16GB 内存。修复：模型加载提到map()外部，作为闭包传入。

4.3 多进程（num_proc）失效诊断：为什么设置了 8 核，CPU 却只有 12% 利用率

num_proc=8不等于 CPU 利用率 800%。常见原因：

原因1：I/O 瓶颈
你的数据在机械硬盘（HDD）上，num_proc=8导致 8 个进程同时读磁盘，反而互相阻塞。
✅ 解决：num_proc=1，或把数据移到 SSD，或用streaming=True+batch_size控制 I/O 压力。

原因2：GIL 锁死
你的map()函数里有大量 Python 字符串操作（如正则、str.replace()），GIL 让多进程无法并行。
✅ 解决：改用 Arrow 计算（pc.utf8_replace_slice()）或numba.jit加速。

原因3：batch_size 太小
batch_size=10时，进程大部分时间花在进程间通信和调度上，而非计算。
✅ 解决：batch_size至少设为num_proc * 100，如num_proc=8→batch_size=800。

快速诊断命令：

# 实时查看进程 CPU 和 I/O htop -p $(pgrep -f "python.*map") # 或用 iostat 看磁盘 iostat -x 1

如果iowait%高，就是 I/O 瓶颈；如果cpu%低且iowait%低，就是 GIL 或 batch_size 问题。

4.4 常见错误速查表

这张表，贴在我工位显示器边框上，三年没换过。它不解决所有问题，但能让你在 30 秒内排除 80% 的低级错误。

5. 生产环境部署建议：从笔记本到 Kubernetes 的平滑迁移

5.1 本地开发 → 云服务器：环境一致性保障

在笔记本上跑通的 pipeline，上云服务器就报错，90% 是环境差异。我的标准化方案：

• Python 环境：用conda env export > environment.yml，而非pip freeze。Conda 能锁定pyarrow、numpy等底层库的 exact version。
• Datasets 版本：团队强制使用datasets==2.16.1（当前最稳定的 LTS 版本），禁用>=。
• 缓存路径：统一设置环境变量export HF_HOME="/data/hf"，所有机器指向同一 NFS 或 EBS 卷，避免重复下载。

# 云服务器初始化脚本 conda env create -f environment.yml conda activate myenv export HF_HOME="/data/hf" mkdir -p $HF_HOME/datasets

5.2 大规模分布式预处理：Kubernetes Job 模板

当数据量超 10TB，单机处理太慢。我们用 K8s Job 分片处理：

# preprocess-job.yaml apiVersion: batch/v1 kind: Job metadata: name: preprocess-part-{{ part_id }} spec: template: spec: containers: - name: preprocessor image: my-registry/preprocess:v1.2 command: ["python", "preprocess.py"] args: - "--data-dir" - "/data/raw" - "--output-dir" - "/data/processed" - "--part-id" - "{{ part_id }}" volumeMounts: - name:>import argparse from datasets import load_dataset def main(): parser = argparse.ArgumentParser() parser.add_argument("--data-dir") parser.add_argument("--output-dir") parser.add_argument("--part-id", type=int) args = parser.parse_args() # 只加载当前分片 part_file = f"{args.data_dir}/part-{args.part_id:05d}.jsonl" # streaming=True 确保内存可控 dataset = load_dataset("json", data_files=[part_file], streaming=True) # 执行你的 map/filter processed = dataset.map(...) # 保存为 arrow，非 json processed.save_to_disk(f"{args.output_dir}/part-{args.part_id}") if __name__ == "__main__": main()

关键经验：

• 每个 Job 处理一个分片，避免锁竞争；
• 输出用save_to_disk()，后续用load_from_disk()直接合并；
• Job 失败时，K8s 会自动重试，无需人工干预。

5.3 监控与告警：让数据 pipeline “可观察”

最后一步，也是最容易被忽视的：给 pipeline 加监控。

• 指标采集：用prometheus_client暴露指标：

  from prometheus_client import Counter, Histogram PROCESSED_SAMPLES = Counter("hf_dataset_processed_samples", "Total samples processed") MAP_DURATION = Histogram("hf_dataset_map_duration_seconds", "Time spent in