Hub documentation

Spark

Hub

API docs

Join the Hugging Face community

and get access to the augmented documentation experience

Collaborate on models, datasets and Spaces

Faster examples with accelerated inference

Switch between documentation themes

to get started

Spark

Spark은 분산 환경에서 실시간 대규모 데이터 처리를 지원합니다.

pyspark_huggingface를 사용하면 “huggingface” Data Source를 통해 PySpark에서 Hugging Face datasets 저장소에 액세스할 수 있습니다.

Hugging Face Spaces에서 Spark Notebooks를 사용해 보세요. PySpark와 pyspark_huggingface가 미리 설치된 Notebooks를 제공합니다.

설정

설치

Hugging Face Datasets에 읽고 쓰기 위해 pyspark_huggingface 라이브러리를 설치해야 합니다:

pip install pyspark_huggingface

이 명령은 인증을 위한 huggingface_hub와 데이터셋 읽기 및 쓰기를 위한 pyarrow와 같은 필요한 종속성도 설치합니다.

인증

Hugging Face에 인증해야 비공개/게이트된 데이터셋 저장소를 읽거나 데이터셋 저장소에 쓸 수 있습니다.

예를 들어 CLI를 사용할 수 있습니다:

hf auth login

HF_TOKEN 환경 변수를 설정하거나 리더에 token 옵션을 전달하여 Hugging Face 토큰을 제공할 수도 있습니다. 인증에 대한 자세한 내용은 이 가이드를 참고하세요.

“huggingface” 데이터 소스 활성화

PySpark 4에는 커스텀 소스의 데이터셋을 사용할 수 있는 새로운 Data Source API가 도입되었습니다. pyspark_huggingface가 설치되어 있으면 PySpark가 이를 자동으로 가져와 “huggingface” Data Source를 활성화합니다.

라이브러리는 PySpark 3.5, 3.4, 3.3에 대해 “huggingface” Data Source의 Data Source API를 백포트합니다. 다만 이 경우 백포트를 활성화하고 “huggingface” Data Source를 사용하려면 pyspark_huggingface를 명시적으로 가져와야 합니다:

>>> import pyspark_huggingface
huggingface datasource enabled for pyspark 3.x.x (backport from pyspark 4)

읽기

“huggingface” Data Source는 pyarrow를 기반으로 Arrow 데이터를 스트리밍하여 Hugging Face에서 데이터셋을 읽을 수 있습니다. 이는 Hugging Face의 지원 형식으로 제공되는 모든 데이터셋, 예를 들어 Parquet 데이터셋과 호환됩니다.

예를 들어 stanfordnlp/imdb 데이터셋을 로드하는 방법은 다음과 같습니다:

>>> import pyspark_huggingface
>>> from pyspark.sql import SparkSession
>>> spark = SparkSession.builder.appName("demo").getOrCreate()
>>> df = spark.read.format("huggingface").load("stanfordnlp/imdb")

다음은 BAAI/Infinity-Instruct 데이터셋을 사용한 또 다른 예시입니다. 이것은 게이트된 저장소이며, 사용자는 액세스하기 전에 이용 약관에 동의해야 합니다. 또한 “3M”과 “7M”이라는 여러 서브셋이 있습니다. 따라서 로드할 서브셋을 지정해야 합니다.

.format() 함수를 사용하여 “huggingface” Data Source를 사용하고, .load()를 사용하여 데이터셋(더 정확히는 7M 샘플을 포함하는 “7M”이라는 이름의 config 또는 subset)을 로드합니다. 그런 다음 언어별 대화 수를 계산하고 데이터셋을 필터링합니다.

게이트된 저장소에 액세스하기 위해 로그인한 후, 다음을 실행할 수 있습니다:

>>> import pyspark_huggingface
>>> from pyspark.sql import SparkSession
>>> spark = SparkSession.builder.appName("demo").getOrCreate()
>>> df = spark.read.format("huggingface").option("config", "7M").load("BAAI/Infinity-Instruct")
>>> df.show()
+---+----------------------------+-----+----------+--------------------+        
| id|               conversations|label|langdetect|              source|
+---+----------------------------+-----+----------+--------------------+
|  0|        [{human, def exti...|     |        en|      code_exercises|
|  1|        [{human, See the ...|     |        en|                flan|
|  2|        [{human, This is ...|     |        en|                flan|
|  3|        [{human, If you d...|     |        en|                flan|
|  4|        [{human, In a Uni...|     |        en|                flan|
|  5|        [{human, Read the...|     |        en|                flan|
|  6|        [{human, You are ...|     |        en|          code_bagel|
|  7|        [{human, I want y...|     |        en|          Subjective|
|  8|        [{human, Given th...|     |        en|                flan|
|  9|[{human, 因果联系原则是法...|     |     zh-cn|          Subjective|
| 10|        [{human, Provide ...|     |        en|self-oss-instruct...|
| 11|        [{human, The univ...|     |        en|                flan|
| 12|        [{human, Q: I am ...|     |        en|                flan|
| 13|        [{human, What is ...|     |        en|      OpenHermes-2.5|
| 14|        [{human, In react...|     |        en|                flan|
| 15|        [{human, Write Py...|     |        en|      code_exercises|
| 16|        [{human, Find the...|     |        en|            MetaMath|
| 17|        [{human, Three of...|     |        en|            MetaMath|
| 18|        [{human, Chandra ...|     |        en|            MetaMath|
| 19|[{human, 用经济学知识分析...|     |     zh-cn|          Subjective|
+---+----------------------------+-----+----------+--------------------+

이것은 데이터셋을 스트리밍 방식으로 로드하며, 출력 DataFrame은 데이터셋의 각 데이터 파일당 하나의 파티션을 갖도록 하여 효율적인 분산 처리를 가능하게 합니다.

언어별 대화 수를 계산하기 위해 columns 옵션과 groupBy() 연산을 사용하는 다음 코드를 실행합니다. columns 옵션은 PySpark가 Data Source API에서 predicate push-down을 지원하지 않으므로 필요한 데이터만 로드하는 데 유용합니다. 특정 범위의 값을 가진 데이터만 로드하기 위한 filters 옵션도 있습니다.

>>> df_langdetect_only = (
...     spark.read.format("huggingface")
...     .option("config", "7M")
...     .option("columns", '["langdetect"]')
...     .load("BAAI/Infinity-Instruct")
... )
>>> df_langdetect_only.groupBy("langdetect").count().show()
+----------+-------+                                                            
|langdetect|  count|
+----------+-------+
|        en|6697793|
|     zh-cn| 751313|
+----------+-------+

데이터셋을 필터링하여 중국어 대화만 유지하려면:

>>> df_chinese_only = (
...     spark.read.format("huggingface")
...     .option("config", "7M")
...     .option("filters", '[("langdetect", "=", "zh-cn")]')
...     .load("BAAI/Infinity-Instruct")
... )
>>> df_chinese_only.show()
+---+----------------------------+-----+----------+----------+                  
| id|               conversations|label|langdetect|    source|
+---+----------------------------+-----+----------+----------+
|  9|[{human, 因果联系原则是法...|     |     zh-cn|Subjective|
| 19|[{human, 用经济学知识分析...|     |     zh-cn|Subjective|
| 38| [{human, 某个考试共有A、...|     |     zh-cn|Subjective|
| 39|[{human, 撰写一篇关于斐波...|     |     zh-cn|Subjective|
| 57|[{human, 总结世界历史上的...|     |     zh-cn|Subjective|
| 61|[{human, 生成一则广告词。...|     |     zh-cn|Subjective|
| 66|[{human, 描述一个有效的团...|     |     zh-cn|Subjective|
| 94|[{human, 如果比利和蒂芙尼...|     |     zh-cn|Subjective|
|102|[{human, 生成一句英文名言...|     |     zh-cn|Subjective|
|106|[{human, 写一封感谢信，感...|     |     zh-cn|Subjective|
|118| [{human, 生成一个故事。}...|     |     zh-cn|Subjective|
|174|[{human, 高胆固醇水平的后...|     |     zh-cn|Subjective|
|180|[{human, 基于以下角色信息...|     |     zh-cn|Subjective|
|192|[{human, 请写一篇文章，概...|     |     zh-cn|Subjective|
|221|[{human, 以诗歌形式表达对...|     |     zh-cn|Subjective|
|228|[{human, 根据给定的指令，...|     |     zh-cn|Subjective|
|236|[{human, 打开一个新的生成...|     |     zh-cn|Subjective|
|260|[{human, 生成一个有关未来...|     |     zh-cn|Subjective|
|268|[{human, 如果有一定数量的...|     |     zh-cn|Subjective|
|273| [{human, 题目：小明有5个...|     |     zh-cn|Subjective|
+---+----------------------------+-----+----------+----------+

로드된 DataFrame에 필터를 적용하거나 컬럼을 제거하는 것도 가능하지만, 특히 Parquet 데이터셋의 경우 로드하는 동안 이를 수행하는 것이 더 효율적입니다. 실제로 Parquet는 파일 및 row group 수준의 메타데이터를 포함하고 있어, 조건을 충족하는 샘플이 없는 데이터셋의 일부 전체를 건너뛸 수 있습니다. Parquet의 컬럼은 독립적으로 로드할 수도 있어, 제외된 컬럼을 건너뛰고 불필요한 데이터를 로드하지 않도록 할 수 있습니다.

옵션

read..option()에 전달할 수 있는 사용 가능한 옵션 목록은 다음과 같습니다:

config (string): 데이터셋의 하위 집합/설정을 선택합니다
split (string): 데이터셋 스플릿을 선택합니다 (기본값은 “train”입니다)
token (string): Hugging Face 토큰

config나 split을 지정하는 대신, 로드할 파일을 수동으로 선택할 수 있습니다:

data_dir (string): 디렉터리 선택
data_files (string): 하나 이상의 파일을 선택합니다. 예: "data/*.parquet" 또는 '["part1.parquet", "par2.parquet"]'

Parquet 데이터셋의 경우:

columns (string): 로드할 컬럼의 하위 집합을 선택합니다. 예: '["id"]'
filters (string): 기준에 맞지 않는 파일과 row group을 건너뛰기 위해 사용합니다. 예: '[("source", "=", "code_exercises")]'. 필터는 pyarrow.parquet.ParquetDataset에 전달됩니다.

기타 옵션은 모두 [datasets.load_dataset] (https://huggingface.co/docs/datasets/en/package_reference/loading_methods#datasets.load_dataset)의 인수로 전달됩니다.

SQL 쿼리 실행

PySpark Dataframe가 준비되면 spark.sql을 사용하여 SQL 쿼리를 실행할 수 있습니다:

>>> import pyspark_huggingface
>>> from pyspark.sql import SparkSession
>>> spark = SparkSession.builder.appName("demo").getOrCreate()
>>> df = (
...     spark.read.format("huggingface")
...     .option("config", "7M")
...     .option("columns", '["source"]')
...     .load("BAAI/Infinity-Instruct")
... )
>>> spark.sql("SELECT source, count(*) AS total FROM {df} GROUP BY source ORDER BY total DESC", df=df).show()
+--------------------+-------+
|              source|  total|
+--------------------+-------+
|                flan|2435840|
|          Subjective|1342427|
|      OpenHermes-2.5| 855478|
|            MetaMath| 690138|
|      code_exercises| 590958|
|Orca-math-word-pr...| 398168|
|          code_bagel| 386649|
|        MathInstruct| 329254|
|python-code-datas...|  88632|
|instructional_cod...|  82920|
|        CodeFeedback|  79513|
|self-oss-instruct...|  50467|
|Evol-Instruct-Cod...|  43354|
|CodeExercise-Pyth...|  27159|
|code_instructions...|  23130|
|  Code-Instruct-700k|  10860|
|Glaive-code-assis...|   9281|
|python_code_instr...|   2581|
|Python-Code-23k-S...|   2297|
+--------------------+-------+

다시 한번 강조하지만, columns 옵션을 지정하는 것은 필수적이지 않지만, 불필요한 데이터를 로드하지 않고 쿼리를 더 빠르게 실행하는 데 유용합니다.

쓰기

“huggingface” Data Source를 사용하여 PySpark Dataframe을 Hugging Face에 작성할 수 있습니다. 이는 Parquet 파일을 분산 방식으로 병렬로 업로드하고, 모든 파일이 업로드된 후에만 파일을 커밋합니다. 동작 방식은 다음과 같습니다:

>>> import pyspark_huggingface
>>> df.write.format("huggingface").save("username/dataset_name")

이 함수를 사용하여 BAAI/Infinity-Instruct 데이터셋의 필터링된 버전을 Hugging Face에 다시 쓰는 방법은 다음과 같습니다.

먼저 데이터셋 저장소를 생성해야 합니다. 예를 들어 username/Infinity-Instruct-Chinese-Only (원하는 경우 비공개로 설정할 수 있습니다). 그 다음, 인증이 되어 있고 “huggingface” Data Source를 사용할 수 있는지 확인한 후, mode를 “overwrite”로 설정합니다(기존 데이터셋을 확장하려면 “append”로 설정). 그리고 .save()를 사용하여 Hugging Face에 push합니다:

>>> df_chinese_only.write.format("huggingface").mode("overwrite").save("username/Infinity-Instruct-Chinese-Only")

모드

Hugging Face에 데이터셋을 푸시할 때 사용할 수 있는 두 가지 모드가 있습니다:

“overwrite”: 이미 존재하는 데이터셋을 덮어쓰기
“append”: 기존 데이터셋에 데이터셋을 추가

옵션

write.option()에 전달할 수 있는 사용 가능한 옵션 목록은 다음과 같습니다:

token (string): Hugging Face 토큰

여기에 더 많은 옵션을 추가하는 기여를 환영합니다. 특히 subset와 split에 대해.

Storage Buckets

It is common to process raw data in Storage Buckets and experiment there before publishing AI-ready data in Dataset repositories.

Dataset 저장소와 동일한 방식으로 Storage Buckets에 액세스하되 buckets/ 접두사와 data_dir 또는 data_files 옵션을 사용합니다:

>>> df = spark.read.format("huggingface").option("data_dir", "data").load("buckets/username/my-bucket")
>>> # OR with a glob pattern
>>> # df = spark.read.format("huggingface").option("data_files", "data/*.parquet").load("buckets/username/my-bucket")
>>> df.write.format("huggingface").option("data_dir", "new-data").save("buckets/username/my-bucket")

Update on GitHub

←성능 최적화 WebDataset→