資料集匯入

資料通常以不適合 AI 工作流程的形式存在於資料庫或雲端儲存中。將資料匯入 Hub 是發布 AI 就緒 (AI-ready) 資料集的好方法，能夠實現輕鬆且高效的資料載入、處理，以及模型訓練與評估。

使用 huggingface_hub

匯入資料最簡單的方式就是直接使用 huggingface_hub 上傳資料檔案。

huggingface_hub Python 函式庫提供了豐富的功能集，讓您可以管理儲存庫，包括建立儲存庫以及將資料集上傳至 Hub。請前往 用戶端函式庫說明文件 以了解更多資訊。

如果您的資料是靜態/凍結的，且您可以輕鬆取得 Hub 支援格式（例如 Parquet 或 JSON Lines）的資料本地副本，並具有可用的結構（例如用於訓練和評估的定義明確的欄位），此方法就很適用。

使用 dlt

dlt 是一個用於資料遷移 (ETL) 的開源 Python 函式庫，對於構建資料管線的開發人員（及其代理）非常有用。它可以從多種來源類型匯入資料：

• 雲端儲存或檔案
• REST API
• SQL 資料庫
• Python 生成器 (Generators)

來源類型範例

• 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

命名空間 (Namespace) 應為您的使用者名稱，或是您想要匯入資料集的組織/團隊名稱。

隨後，每個 dlt 資料集都會建立或更新一個 Hugging Face 資料集儲存庫。儲存庫名稱為 <namespace>/<dataset_name>，其中 <namespace> 與您在 bucket_url 中使用的相同（您的組織或團隊），而 <dataset_name> 則是管線的 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)。

排程匯入

在 Hub 上頻繁更新同一個檔案會面臨一些限制。例如，您可能想要匯入正在執行的 LLM 推論伺服器的生成結果、即時代理追蹤記錄，或正在進行的模型訓練日誌。在這種情況下，將資料作為 Hub 上的資料集上傳是合理的，但要妥善處理可能會很困難。主要原因在於您不希望對資料的每次更新都進行版本控制，因為這會使 Git 儲存庫變得無法使用。

有三種可用選項：

• 使用儲存桶 (Storage Bucket) 而非資料集儲存庫： 儲存桶 提供類似 S3 的體驗，允許非常頻繁地更新檔案，因為它們不基於 Git。儲存桶對於尚未準備好發布為資料集的資料特別有用，例如仍在演進中或需要進一步整理的資料。
• 使用 CommitScheduler： huggingface_hub 中的 CommitScheduler 提供接近即時的匯入功能，以保持資料集儲存庫的 Git 歷史紀錄易於管理。它可以設定為以分鐘為單位的間隔執行 Git 提交。
• 使用 Hugging Face Jobs 排程匯入腳本： Hugging Face Jobs 提供了一種在 Hugging Face 基礎設施上執行和排程 Python 腳本的方法。您可以設定匯入腳本以使用 Cron 語法定義的間隔執行。

使用儲存桶進行高頻匯入

與基於 Git 的資料集儲存庫不同，您可以以極高的頻率更新儲存桶中的檔案，從而實現近乎即時的匯入。

使用 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)])

或者，您可以附加檔案到儲存桶中，並在每個新項目上執行 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，其預設區塊大小為 5MiB，這意味著執行 flush 時，會在填滿 5MiB 的新資料後才進行上傳。如果您希望更頻繁地上傳，請在 hffs.open() 中降低 blocksize（例如 hffs.open(..., blocksize=100 * 2 ** 10) 即為 100 kiB）或是使用 f.flush(force=True)。

Hugging Face 儲存基於 Xet，能夠在附加檔案時實現高效的 I/O：上傳內容會進行去重 (deduplicated)，僅上傳新資料。關於在儲存桶中執行動態資料匯入的更多資訊，請查看 儲存桶上傳說明文件 以及 資料集編輯說明文件。

使用 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 說明文件。

使用 Hugging Face Jobs 進行 Cron 風格的排程

排程 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 說明文件。