摄取数据集

数据通常以不适合 AI 工作流的形式存储在数据库或云存储中。 将数据摄取到 Hub 是将其发布为 AI 就绪数据集的好方法，从而实现轻松高效的数据加载、处理以及模型训练和评估。

使用 huggingface_hub

最简单的方式是使用 huggingface_hub 直接上传数据文件来导入数据。

huggingface_hub Python 库提供了一套丰富的功能，可让您管理仓库，包括创建仓库以及向 Hub 上传数据集。请访问 客户端库文档 以了解更多信息。

如果您的数据是静态/固定的，并且您可以轻松获取该数据的本地副本，且其格式为 Hub 所支持（例如 Parquet 或 JSON Lines），结构也便于使用（例如，用于训练和评估的字段定义明确），那么这一点就很重要。

使用 dlt

dlt 是一个用于数据移动（ETL）的开源 Python 库，对于构建数据管道的开发者（及其智能体）非常有用。 它可以摄取来自多种来源类型的数据：

• 云存储或文件
• REST API
• SQL 数据库
• Python 生成器

源类型示例：

• filesystem（包括 s3、gs、az、abff 等）
• sql_database, mongodb, google_sheets
• notion, hubspot, rest_api

从数据源列表中找到您的数据源类型，并创建您的dlt项目：

dlt init <source-type> filesystem

然后，您可以在 dlt 项目的根目录中创建一个配置文件 .dlt/secrets.toml，基于 hf:// 协议将 Hub 定义为您数据集的文件系统目标：

[destination.filesystem]
bucket_url = "hf://datasets/<namespace>"

[destination.filesystem.credentials]
hf_token = "hf_..."  # Your Hugging Face Access Token

命名空间应为您希望导入数据集的用户名或组织/团队名称。

然后，每个 dlt 数据集都会创建或更新一个 Hugging Face 数据集仓库。仓库名称为 <namespace>/<dataset_name>，其中 <namespace> 与您在 bucket_url 中使用的相同（即您的组织或团队），而 <dataset_name> 则是该 pipeline 的 dataset_name。

以下是一个示例流水线：

import dlt

@dlt.resource
def my_data():
    # One of the functions auto-generated by `dlt init` that you can customize,
    # or you can define your own python generator function.

    # Here is an example from the `chess` source type:
    for player in ['magnuscarlsen', 'rpragchess']:
        response = requests.get(f'https://api.chess.com/pub/player/{player}')
        response.raise_for_status()
        yield response.json()

# Requires bucket_url = "hf://datasets/<namespace>" in .dlt/secrets.toml
pipeline = dlt.pipeline(
    pipeline_name="my_pipeline",
    destination="filesystem",
    dataset_name="dataset_name",
)

pipeline.run(my_data())

自定义 dlt 资源以加载所需的数据，并解析要在数据集中发布的字段，例如训练和评估所需的文本。

使用其他库

一些库，例如 🤗 Datasets、Pandas、Polars、Dask、DuckDB、Spark 或 Daft，可以从各种来源将数据摄取到 Hub 中。 有关更多信息，请参阅 Datasets Hub 支持的库列表。

摄取原始数据

如果您正在摄取需要进一步整理才能发布为 AI 就绪数据集的原始数据，或者您需要类似 S3 的体验，请考虑将它们摄取到 Hugging Face Storage Buckets。如需将本地磁盘或另一云上的大型现有集合迁移到存储桶中，请参阅 Migration Guide。

如果您的数据已经存在于数据集、模型或 Space 仓库中，您可以使用 hf buckets cp 将其 在服务端 复制到存储桶中——无需下载或重新上传：

hf cp hf://datasets/username/my-dataset/ hf://buckets/username/my-bucket/

仅迁移 Xet 内容哈希，因此即使非常大的文件也能瞬间完成复制。请参阅 在仓库和存储桶之间复制文件。

计划摄取

在 Hub 上对同一文件进行数千次更新时存在一些限制。 例如，您可能希望摄取正在运行的 LLM 推理服务器的生成结果、实时智能体的轨迹，或正在运行的模型训练日志。 在这种情况下，将数据作为数据集上传到 Hub 是合理的，但正确执行起来可能比较困难。 主要原因是，您不希望为数据的每次更新都创建版本，因为这会使 git 仓库变得无法使用。

有三种可选方案：

• 使用存储桶（Storage Bucket）而非数据集仓库： 存储桶提供类似 S3 的体验，允许非常频繁地更新文件，因为它们不基于 git。存储桶尤其适用于尚未准备好作为数据集发布的数据，例如仍在演变中或需要更多整理的数据。
• 使用 CommitScheduler：huggingface_hub 中的 CommitScheduler 提供近实时的数据摄取功能，以便将 Dataset 仓库的 git 历史记录保持在可管理的范围内。它可以配置为按以分钟为单位的间隔执行 git 提交。
• 使用 Hugging Face Jobs 来调度摄取脚本：Hugging Face Jobs 提供了一种在 Hugging Face 基础设施上运行和调度 Python 脚本的方式。您可以使用 Cron 语法定义的时间间隔来调度摄取脚本的运行。

高频使用存储桶

与基于 git 的 Dataset 仓库不同，您可以以极高的速率更新 Storage Buckets 上的文件，从而实现准实时数据摄取。

使用 huggingface_hub 中的 batch_bucket_files() 来更新桶中的文件：

from huggingface_hub import batch_bucket_files

def update_bucket(local_files):
    destinations = [os.path.basename(local_file) for local_file in local_file]
    batch_bucket_files(bucket_id="username/bucket_name", add=[(local_file, dst) for local_file, dst in zip(local_files, destinations)])

或者，您可以向 Bucket 中的文件追加内容，并在每个新条目时调用 flush()：

from huggingface_hub import hffs

with hffs.open("buckets/username/bucket_name/texts.jsonl", "a") as f:
    for text in live_texts_stream:
        f.write(json.dumps({"text": text}) + "\n")
        f.flush()

HfFileSystem 基于 fsspec，其默认 blocksize 为 5MiB，这意味着只有当追加的新数据达到完整的 5MiB 块时，flush 操作才会实际上传数据。 如果您希望更频繁地上传，请在 hffs.open() 中调低 blocksize（例如 hffs.open(..., blocksize=100 * 2 ** 10) 表示 100 kiB），或使用 f.flush(force=True)。

Hugging Face 存储基于 Xet，在追加文件时可实现高效的 I/O：上传内容会进行去重，仅上传新增数据。 有关在存储桶中执行动态数据摄入的更多信息，请参阅 存储桶文档中的上传部分 以及 数据集编辑文档。

使用 CommitScheduler 实现近实时

其思路是运行一个后台任务，定期将本地文件夹推送到 Hub。你需要将数据保存到 Hub（可能包含数百万条记录），但无需实时保存每个用户的输入。相反，你可以将数据保存在本地的 JSON 文件中，并每 10 分钟上传一次。例如：

import json
from huggingface_hub import CommitScheduler

folder_path = "path/to/files/to/ingest"
every = 10  # ingest every 10min

with CommitScheduler(repo_id="username/dataset_name", repo_type="dataset", folder_path=folder_path, every=every) as scheduler:
    # Write to the folder to ingest every 10min
    # For example:
    with open(folder_path + "/texts.jsonl", "a") as f:
        f.write(json.dumps({"text": text}) + "\n")
    ...

有关如何在不每次重新上传所有内容的情况下摄取动态数据，请参阅数据集编辑文档。

有关计划上传的更多信息，请参阅 huggingface_hub 文档。

基于 Cron 的 Hugging Face Jobs

按计划调度 Python 脚本以摄取数据

例如，要每 5 分钟运行一次脚本 ingest.py：

hf jobs scheduled uv run "*/5 * * * *" ingest.py

在脚本头部声明脚本依赖项，或使用--with。 例如，要在每天午夜运行dlt管道：

hf jobs scheduled uv run --with "dlt[hf]" "0 0 * * *" pipeline.py

您可以使用 hf jobs logs 或在 Hugging Face 账户的 Jobs 页面上直接查看每次运行的日志。

有关 Hugging Face Jobs 的更多信息，请参阅 Jobs 文档。