# Spark

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

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

Hugging Face Spaces에서 [Spark Notebooks](https://huggingface.co/spaces/Dataset-Tools/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 토큰을 제공할 수도 있습니다. 인증에 대한 자세한 내용은 [이 가이드](https://huggingface.co/docs/huggingface_hub/quick-start#authentication)를 참고하세요.

### "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`를 명시적으로 가져와야 합니다:

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

## 읽기

"huggingface" Data Source는 `pyarrow`를 기반으로 Arrow 데이터를 스트리밍하여 Hugging Face에서 데이터셋을 읽을 수 있습니다. 이는 Hugging Face의 [지원 형식](https://huggingface.co/docs/hub/datasets-adding#file-formats)으로 제공되는 모든 데이터셋, 예를 들어 Parquet 데이터셋과 호환됩니다.

예를 들어 [stanfordnlp/imdb](https://huggingface.co/stanfordnlp/imdb) 데이터셋을 로드하는 방법은 다음과 같습니다:

```python
>>> 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](https://huggingface.co/datasets/BAAI/Infinity-Instruct) 데이터셋을 사용한 또 다른 예시입니다. 이는 게이트된 저장소로, 사용자는 액세스하기 전에 이용 약관에 동의해야 합니다. 또한 "3M"과 "7M"이라는 여러 서브셋이 있습니다. 따라서 로드할 서브셋을 지정해야 합니다.

    
    

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

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

```python
>>> 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()` 연산을 사용하는 다음 코드를 실행합니다. PySpark는 Data Source API에서 predicate push-down을 지원하지 않으므로, 필요한 데이터만 로드하기 위해 `columns` 옵션이 유용합니다. 특정 범위의 값을 가진 데이터만 로드하기 위한 `filters` 옵션도 있습니다.

```python
>>> 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|
+----------+-------+
```

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

```python
>>> 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](https://arrow.apache.org/docs/python/generated/pyarrow.parquet.ParquetDataset.html)에 전달됩니다.

기타 옵션은 모두 [datasets.load_dataset]의 인수로 전달됩니다.

### SQL 쿼리 실행

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

```python
>>> 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 파일을 분산 방식으로 병렬로 업로드하고, 모든 파일이 업로드된 후에만 파일을 커밋합니다. 동작 방식은 다음과 같습니다:

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

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

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

```python
>>> 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

[Storage Buckets](/docs/hub/storage-buckets)에서 원시 데이터를 처리하고, Dataset 저장소에 AI에 적합한 데이터를 게시하기 전에 해당 환경에서 실험하는 것이 일반적입니다.

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

```python
>>> 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")
```