データセットの取り込み

データは通常、データベースやクラウドストレージに、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

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へのインジェストを検討してください。ローカルディスクや他のクラウドから既存の大量のコレクションをバケットに移動するには、Migration Guideを参照してください。

データがすでに dataset、model、または Space リポジトリに存在する場合、hf buckets cp を使用して、ダウンロードや再アップロードなしにバケットへサーバーサイドでコピーできます:

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

Xetのコンテンツハッシュのみが移行されるため、非常に大きなファイルでも即座にコピーされます。リポジトリとバケット間のファイルのコピーを参照してください。

スケジュールされた取り込み

Hub上で同じファイルを数千回更新する際には、いくつかの制限があります。 例えば、稼働中のLLM推論サーバーの生成結果、ライブエージェントのトレース、または稼働中のモデルトレーニングのログを取り込みたい場合があります。 そのような場合、データをHub上のデータセットとしてアップロードするのが理にかなっていますが、適切に行うのは難しい場合があります。 主な理由は、データのすべての更新をバージョン管理するとgitリポジトリが使用不能になるため、そうしたくないからです。

3つのオプションが利用可能です:

• Dataset リポジトリの代わりに Storage Bucket を使用する: Storage Bucket は git をベースとしないため、ファイルを非常に高頻度で更新できる S3 同様の体験を提供します。Storage Bucket は、Dataset として公開する準備ができていないデータ、例えばまだ進化しているデータやさらなるキュレーションが必要なデータに特に有用です。
• CommitSchedulerの使用: huggingface_hubのCommitSchedulerは、Datasetリポジトリのgit履歴を管理可能な状態に保つための、ほぼリアルタイムの取り込みを提供します。分単位で定義された間隔でgitコミットを行うように設定できます。
• Hugging Face Jobs を使用して取り込みスクリプトのスケジュールを設定する: Hugging Face Jobs は、Hugging Face のインフラ上で Python スクリプトを実行およびスケジュールするための方法を提供します。Cron 構文で定義された間隔で、取り込みスクリプトの実行をスケジュールできます。

Storage Buckets を使用した高頻度

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 をベースとしており、デフォルトのブロックサイズは 5MiB です。つまり、フラッシュを実行すると、5MiB の新しいデータが 1 つのチャンク分追加された時点で、実際にデータがアップロードされます。 より頻繁にアップロードしたい場合は、hffs.open() の blocksize を下げてください（例: 100 kiB の場合は hffs.open(..., blocksize=100 * 2 ** 10) ）。または、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のドキュメントを参照してください。

Hugging Face Jobs を使用した Cron ベース

スケジュールに従ってデータをインジェストするよう、Pythonスクリプトをスケジュールします

例えば、ingest.py スクリプトを5分ごとに実行するには:

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のドキュメントを参照してください。