# Spark

Spark, dağıtık bir ortamda gerçek zamanlı ve büyük ölçekli veri işleme imkânı sağlar.

`pyspark_huggingface` aracını kullanarak PySpark'ta "huggingface" Veri Kaynağı üzerinden Hugging Face veri seti depolarına erişebilirsiniz.

Hugging Face Spaces'te [Spark Notebooks](https://huggingface.co/spaces/Dataset-Tools/Spark-Notebooks)'u deneyin; PySpark ve `pyspark_huggingface` önceden yüklü olarak gelen Notebooks'a erişin.

## Kurulum

### Kurulum

Hugging Face Datasets'e okuma ve yazma yapabilmek için `pyspark_huggingface` kütüphanesini yüklemeniz gerekir:

```
pip install pyspark_huggingface
```

Bu, kimlik doğrulama için `huggingface_hub` ve veri kümelerini okuma/yazma için `pyarrow` gibi gerekli bağımlılıkları da yükleyecektir.

### Kimlik Doğrulama

Özel/korumalı veri seti depolarını okumak veya veri seti depolarınıza yazmak için Hugging Face'e kimlik doğrulamanız gerekir.

Örneğin CLI'yi şu şekilde kullanabilirsiniz:

```
hf auth login
```

Hugging Face token'ınızı `HF_TOKEN` ortam değişkeni ile veya okuyucuya `token` seçeneğini geçirerek de sağlayabilirsiniz. Kimlik doğrulama hakkında daha fazla ayrıntı için [bu kılavuza](https://huggingface.co/docs/huggingface_hub/quick-start#authentication) göz atın.

### "huggingface" Veri Kaynağını Etkinleştirin

PySpark 4, özel kaynaklardan veri setleri kullanmayı sağlayan yeni bir Data Source API ile birlikte geldi. `pyspark_huggingface` yüklüyse, PySpark bunu otomatik olarak içe aktarır ve "huggingface" Data Source'unu etkinleştirir.

Kütüphane, PySpark 3.5, 3.4 ve 3.3 için "huggingface" Veri Kaynağı'nın Data Source API'sini geriye dönük olarak da destekler. Ancak bu durumda, geriye dönük desteği etkinleştirmek ve "huggingface" Veri Kaynağı'nı kullanıma açmak için `pyspark_huggingface` modülünün açıkça içe aktarılması gerekir:

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

## Oku

"huggingface" Veri Kaynağı, `pyarrow` kullanarak Arrow verilerini akış olarak okumak için Hugging Face'ten veri setlerini okumaya olanak tanır. Bu, Hugging Face'teki [desteklenen biçimlerdeki](https://huggingface.co/docs/hub/datasets-adding#file-formats) tüm veri setleriyle, örneğin Parquet veri setleriyle uyumludur.

Örneğin [stanfordnlp/imdb](https://huggingface.co/stanfordnlp/imdb) veri setinin nasıl yükleneceği aşağıdaki gibidir:

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

İşte [BAAI/Infinity-Instruct](https://huggingface.co/datasets/BAAI/Infinity-Instruct) veri setiyle ilgili başka bir örnek. Bu, kapalı bir depodur; kullanıcıların erişmeden önce kullanım koşullarını kabul etmesi gerekir. Ayrıca "3M" ve "7M" adlı birden fazla alt kümesi vardır. Bu nedenle yükleyeceğimiz alt kümenin hangisi olduğunu belirtmemiz gerekir.

    
    

veri kaynağını kullanmak için `.format()` fonksiyonunu, veri setini (daha doğrusu 7M örnek içeren "7M" adlı yapılandırma veya alt kümesini) yüklemek için ise `.load()` fonksiyonunu kullanırız. Ardından dil başına diyalog sayısını hesaplar ve veri setini filtreleriz.

Korumalı depoya erişmek için oturum açtıktan sonra şunu çalıştırabiliriz:

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

Bu, veri setini akış biçiminde yükler ve çıktı DataFrame'i, verimli dağıtık işleme imkânı sağlamak üzere veri setindeki her veri dosyası için bir bölüme sahiptir.

Dil başına diyalog sayısını hesaplamak için `columns` seçeneğini ve bir `groupBy()` işlemini kullanan bu kodu çalıştırıyoruz. PySpark, Data Source API ile öngörü itme (predicate push-down) özelliğini etkinleştirmediğinden, yalnızca ihtiyaç duyduğumuz verileri yüklemek için `columns` seçeneği faydalıdır. Belirli bir aralıkta değerlere sahip verileri yüklemek için ayrıca bir `filters` seçeneği de mevcuttur.

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

Veri setini filtrelemek ve yalnızca Çince diyalogları tutmak için:

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

Yüklenen DataFrame üzerinde filtre uygulamak veya sütunları kaldırmak da mümkündür, ancak bunu özellikle Parquet veri kümelerinde yükleme sırasında yapmak daha verimlidir. Nitekim Parquet, dosya ve satır grubu düzeyinde meta veri içerir; bu da kriterleri karşılayan örnekler içermeyen veri kümesinin tüm bölümlerinin atlanmasına olanak tanır. Parquet'teki sütunlar bağımsız olarak yüklenebilir; bu da hariç tutulan sütunların atlanmasına ve gereksiz verilerin yüklenmemesine olanak tanır.

### Seçenekler

`read..option()` fonksiyonuna iletebileceğiniz mevcut seçeneklerin listesi aşağıdadır:

* `config` (string): bir veri kümesi alt kümesini/yapılandırmasını seçin
* `split` (string): bir veri seti bölümü seçin (varsayılan "train")
* `token` (string): Hugging Face token'ınız

Bir yapılandırma veya bölme belirtmek yerine, yüklemek istediğiniz dosyaları manuel olarak seçebilirsiniz:

* `data_dir` (string): bir dizin seçin
* `data_files` (string): bir veya birden fazla dosya seçin, örn. `"data/*.parquet"` veya `'["part1.parquet", "par2.parquet"]'`

Parquet veri kümeleri için:

* `columns` (string): yüklemek için sütunların bir alt kümesini seçin, örn. `'["id"]'`
* `filters` (string): bir kriterle eşleşmeyen dosyaları ve satır gruplarını atlamak için kullanılır, örn. `'[("source", "=", "code_exercises")]'`. Filtreler [pyarrow.parquet.ParquetDataset](https://arrow.apache.org/docs/python/generated/pyarrow.parquet.ParquetDataset.html)'e iletilir.

Diğer tüm seçenekler, [datasets.load_dataset] (https://huggingface.co/docs/datasets/en/package_reference/loading_methods#datasets.load_dataset) fonksiyonuna bir argüman olarak iletilir.

### SQL sorguları çalıştırın

PySpark Dataframe'inizi hazırladığınızda, `spark.sql` kullanarak SQL sorguları çalıştırabilirsiniz:

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

Yine, `columns` seçeneğini belirtmek zorunlu değildir ancak gereksiz verilerin yüklenmesini önleyerek sorguyu hızlandırmak için faydalıdır.

## Yaz

PySpark Dataframe'ını "huggingface" Data Source'u ile Hugging Face'e yazabilirsiniz. Bu kaynak, Parquet dosyalarını dağıtık bir şekilde paralel olarak yükler ve dosyaların tamamı yüklendikten sonra yalnızca dosyaları kaydeder. İşleyişi şu şekildedir:

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

Bu fonksiyonu kullanarak [BAAI/Infinity-Instruct](https://huggingface.co/datasets/BAAI/Infinity-Instruct) veri setinin filtrelenmiş sürümünü Hugging Face'e nasıl yazabileceğimizi aşağıda görebilirsiniz.

Önce bir [veri seti deposu oluşturmanız](https://huggingface.co/new-dataset) gerekir, örneğin `username/Infinity-Instruct-Chinese-Only` (isterseniz özel olarak ayarlayabilirsiniz). Ardından, kimlik doğrulamanızın yapıldığından ve "huggingface" Veri Kaynağı'nı kullanabildiğinizden emin olun, `mode` değerini "overwrite" olarak ayarlayın (mevcut bir veri setini genişletmek istiyorsanız "append") ve `.save()` ile Hugging Face'e gönderin:

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

    
    

### Mod

Hugging Face'e bir veri seti gönderirken iki mod kullanılabilir:

* "overwrite": zaten var ise veri setini üzerine yaz
* "append": veri setini mevcut bir veri setine ekleyin

### Seçenekler

`write.option()` fonksiyonuna iletebileceğiniz mevcut seçeneklerin listesi aşağıdadır:

* `token` (string): Hugging Face token'ınız

Buraya daha fazla seçenek eklemek için katkılar memnuniyetle karşılanır, özellikle `subset` ve `split`.

## Depolama Kovaları

Ham verileri [Storage Buckets](/docs/hub/storage-buckets) içinde işlemek ve AI'ye hazır verileri Dataset depolarında yayınlamadan önce orada deney yapmak yaygın bir uygulamadır.

Depo kovalarını (Storage Buckets), Veri seti depolarıyla aynı şekilde ancak `buckets/` öneki ve `data_dir` veya `data_files` seçenekleriyle erişin:

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