存储桶数据优化实操：10分钟极速管道处理

利用存储桶优化数据管道工作流

新媒网跨境获悉，传统的跨境数据处理，往往面临效率低下、操作复杂等问题。以下是一种更优化的方案，可以帮助中国的跨境团队提升工作效率、节约成本，同时保持数据的完整性和灵活性。

传统做法的挑战：下载、合并、重上传

在以往，许多数据管道工作需要每隔数小时从整个数据库下载全部数据（动辄几个GB），然后再通过 API 获取新增数据或更新记录。接下来，我们需要合并新数据和现存数据，确保去重后将其重新上传。虽然这种方法可以完成任务，但每次都要浪费大量带宽和硬盘空间。此外，如果在工作流程中出了问题，还可能导致重复操作或数据丢失，显然不是长久之策。

新思路：存储桶作为活跃层

新媒网跨境了解到，存储桶是一种可修改且不记录版本的对象储存技术，与传统数据集仓库不同，它没有复杂的版本控制逻辑，只提供简单自由的读写、覆盖和同步能力。其底层技术使用 Xet，通过数据分块及文件去重提高效率，特别适合增量式的自动化流程。

当数据被分块处理后，只有新增或改变部分需要传输，从而显著提高了同步速度，降低了数据冗余。这使得存储桶成为数据管道的最佳工作层，通过以下结构实现数据高效处理：

[采集任务] → [存储桶] → [汇总处理任务] → [发布到数据集仓库]
- （频繁）（活跃存储）（每日）（稳定发布）

在这种模式下，数据采集任务会持续进行，将新数据追加到存储桶中，采集失败也不会影响已存数据。存储桶充当了原始数据的唯一可信来源，数据积累高效且成本低。随后，汇总任务每日运行，对存储桶中的数据进行清理和去重处理，最终发布到版本化的数据仓库。不同阶段互不影响，每步只向前推进，确保流程的稳定性。

实践中的工作流程

数据采集：批量追加到 JSONL 文件

在采集任务中，每次运行都会将新数据记录保存到一个 JSONL 文件中，并直接写入存储桶：

from huggingface_hub import batch_bucket_files

jsonl_data = "\n".join(record.model_dump_json() for record in results)

batch_bucket_files(
    "my-org/my-pipeline-bucket", 
    add=[(jsonl_data.encode(), f"data/{timestamp}.jsonl")],
)

这一过程中无需读取现有数据，也不需要合并步骤，只需简单追加即可。

数据汇总：同步、处理及发布

每日汇总任务会把存储桶中的数据下载到本地并使用 Polars 进行数据处理，最后上传到数据集仓库：

from huggingface_hub import sync_bucket
import polars as pl
from datasets import Dataset

# 从存储桶同步数据
sync_bucket(
    source="hf://buckets/my-org/my-pipeline-bucket/data", 
    dest=local_dir,
)

# Polars LazyFrame: 一次性处理过滤与去重
lf = pl.concat([pl.scan_ndjson(f) for f in jsonl_files])
lf = lf.sort("last_modified", descending=True)
lf = lf.unique(subset=["id"], keep="first")
df = lf.collect()

Dataset.from_polars(df).push_to_hub("my-org/my-dataset")

通过这种方法，可以快速高效处理几十万条记录，保持数据质量。

自动化调度：HF Jobs 定时任务

通过 HF Jobs，企业团队可以依赖 Hugging Face 的云基础设施，按秒计费，并以 cron 风格完成定时任务设置，无需额外手动操作，如：

每隔两小时自动执行采集任务
每日深夜固定时间运行汇总处理

具体调用方式可以参考以下脚本：

# 采集任务（两小时一次）
hf jobs scheduled uv run "8 */2 * * *" \
  -s HF_TOKEN --timeout 30m \
  https://huggingface.co/datasets/my-org/my-pipeline/resolve/main/fetch.py

# 汇总任务（每日固定时间）
hf jobs scheduled uv run "3 3 * * *" \
  -s HF_TOKEN --flavor cpu-upgrade --timeout 1h \
  https://huggingface.co/datasets/my-org/my-pipeline/resolve/main/compile.py

几大优势：