# Spark

Spark memungkinkan pemrosesan data skala besar secara real-time dalam lingkungan terdistribusi.

Anda dapat menggunakan `pyspark_huggingface` untuk mengakses repositori dataset Hugging Face di PySpark melalui Data Source "huggingface".

Coba [Spark Notebooks](https://huggingface.co/spaces/Dataset-Tools/Spark-Notebooks) di Hugging Face Spaces untuk mendapatkan Notebooks dengan PySpark dan `pyspark_huggingface` yang sudah terpasang.

## Persiapan

### Instalasi

Untuk dapat membaca dan menulis ke Hugging Face Datasets, Anda perlu menginstal pustaka `pyspark_huggingface`:

```
pip install pyspark_huggingface
```

Ini juga akan menginstal dependensi yang diperlukan seperti `huggingface_hub` untuk autentikasi, dan `pyarrow` untuk membaca dan menulis dataset.

### Autentikasi

Anda perlu melakukan autentikasi ke Hugging Face untuk membaca repositori dataset privat/terbatas atau untuk menulis ke repositori dataset Anda.

Anda dapat menggunakan CLI, misalnya:

```
hf auth login
```

Anda juga dapat menyediakan token Hugging Face Anda dengan variabel lingkungan `HF_TOKEN` atau dengan meneruskan opsi `token` ke pembaca. Untuk detail lebih lanjut tentang autentikasi, lihat [panduan ini](https://huggingface.co/docs/huggingface_hub/quick-start#authentication).

### Aktifkan Sumber Data "huggingface"

PySpark 4 hadir dengan Data Source API baru yang memungkinkan penggunaan dataset dari sumber kustom. Jika `pyspark_huggingface` terinstal, PySpark akan mengimpor secara otomatis dan mengaktifkan Data Source "huggingface".

Perpustakaan ini juga melakukan backport API Data Source untuk Data Source "huggingface" untuk PySpark 3.5, 3.4, dan 3.3. Namun, dalam kasus ini `pyspark_huggingface` harus diimpor secara eksplisit untuk mengaktifkan backport dan memungkinkan Data Source "huggingface":

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

## Baca

Data Source "huggingface" memungkinkan pembacaan dataset dari Hugging Face, menggunakan `pyarrow` di baliknya untuk streaming data Arrow. Ini kompatibel dengan semua dataset dalam [format yang didukung](https://huggingface.co/docs/hub/datasets-adding#file-formats) di Hugging Face, seperti dataset Parquet.

Sebagai contoh, berikut adalah cara memuat dataset [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")
```

Berikut adalah contoh lain dengan dataset [BAAI/Infinity-Instruct](https://huggingface.co/datasets/BAAI/Infinity-Instruct). Ini adalah repositori bergerbang, pengguna harus menerima syarat penggunaan sebelum mengaksesnya. Dataset ini juga memiliki beberapa subset, yaitu "3M" dan "7M". Jadi, kita perlu menentukan subset mana yang akan dimuat.

    
    

Kami menggunakan fungsi `.format()` untuk memakai Data Source "huggingface", dan `.load()` untuk memuat dataset (lebih tepatnya config atau subset bernama "7M" yang berisi 7M sampel). Kemudian kami menghitung jumlah dialog per bahasa dan memfilter dataset.

Setelah masuk untuk mengakses repositori yang dibatasi, kita dapat menjalankan:

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

Ini memuat dataset dengan cara streaming, dan DataFrame output memiliki satu partisi per file data dalam dataset untuk memungkinkan pemrosesan terdistribusi yang efisien.

Untuk menghitung jumlah dialog per bahasa, kami menjalankan kode ini yang menggunakan opsi `columns` dan operasi `groupBy()`. Opsi `columns` berguna untuk hanya memuat data yang kami butuhkan, karena PySpark tidak mengaktifkan predicate push-down dengan Data Source API. Ada juga opsi `filters` untuk hanya memuat data dengan nilai dalam rentang tertentu.

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

Untuk memfilter dataset dan hanya menyimpan dialog dalam bahasa Mandarin:

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

Memungkinkan juga untuk menerapkan filter atau menghapus kolom pada DataFrame yang dimuat, tetapi lebih efisien melakukannya saat proses pemuatan, terutama pada dataset Parquet. Memang, Parquet berisi metadata pada level file dan row group, yang memungkinkan untuk melewati seluruh bagian dataset yang tidak berisi sampel yang memenuhi kriteria. Kolom dalam Parquet juga dapat dimuat secara independen, yang memungkinkan untuk melewati kolom yang dikecualikan dan menghindari pemuatan data yang tidak diperlukan.

### Opsi

Berikut adalah daftar opsi yang tersedia yang dapat Anda teruskan ke `read..option()`:

* `config` (string): pilih subset/konfigurasi dataset
* `split` (string): pilih split dataset (default adalah "train")
* `token` (string): token Hugging Face Anda

Alih-alih menentukan config atau split, Anda dapat memilih file yang akan dimuat secara manual:

* `data_dir` (string): pilih direktori
* `data_files` (string): pilih satu atau banyak file, misalnya `"data/*.parquet"` atau `'["part1.parquet", "par2.parquet"]'`

Untuk dataset Parquet:

* `columns` (string): pilih subset kolom untuk dimuat, misalnya `'["id"]'`
* `filters` (string): untuk melewati file dan grup baris yang tidak sesuai dengan kriteria, misalnya `'[("source", "=", "code_exercises")]'`. Filter diteruskan ke [pyarrow.parquet.ParquetDataset](https://arrow.apache.org/docs/python/generated/pyarrow.parquet.ParquetDataset.html).

Semua opsi lain diteruskan sebagai argumen ke [datasets.load_dataset] (https://huggingface.co/docs/datasets/en/package_reference/loading_methods#datasets.load_dataset)

### Jalankan kueri SQL

Setelah Anda menyiapkan PySpark Dataframe, Anda dapat menjalankan kueri SQL menggunakan `spark.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|
+--------------------+-------+
```

Sekali lagi, menentukan opsi `columns` tidak diperlukan, tetapi berguna untuk menghindari pemuatan data yang tidak perlu dan membuat kueri lebih cepat.

## Tulis

Anda dapat menulis PySpark Dataframe ke Hugging Face dengan Data Source "huggingface". Sumber data ini mengunggah file Parquet secara paralel dengan cara terdistribusi, dan hanya mengonfirmasi file setelah semuanya terunggah. Cara kerjanya:

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

Berikut adalah cara kita dapat menggunakan fungsi ini untuk menulis kembali versi terfilter dari dataset [BAAI/Infinity-Instruct](https://huggingface.co/datasets/BAAI/Infinity-Instruct) ke Hugging Face.

Pertama, Anda perlu [membuat repositori dataset](https://huggingface.co/new-dataset), misalnya `username/Infinity-Instruct-Chinese-Only` (Anda dapat menjadikannya privat jika diinginkan). Kemudian, pastikan Anda telah terautentikasi dan dapat menggunakan Sumber Data "huggingface", atur `mode` menjadi "overwrite" (atau "append" jika Anda ingin memperluas dataset yang sudah ada), dan dorong ke Hugging Face dengan `.save()`:

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

    
    

### Mode

Dua mode tersedia saat mendorong dataset ke Hugging Face:

* "overwrite": timpa dataset jika sudah ada
* "append": tambahkan dataset ke dataset yang sudah ada

### Opsi

Berikut adalah daftar opsi yang tersedia yang dapat Anda teruskan ke `write.option()`:

* `token` (string): token Hugging Face Anda

Kontribusi untuk menambahkan lebih banyak opsi di sini sangat diterima, khususnya `subset` dan `split`.

## Storage Buckets

Umumnya, data mentah diproses di [Storage Buckets](/docs/hub/storage-buckets) dan diuji coba di sana sebelum menerbitkan data yang siap untuk AI di repositori Dataset.

Akses Storage Buckets dengan cara yang sama seperti repositori Dataset, tetapi dengan prefiks `buckets/` dan opsi `data_dir` atau `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")
```