Spark

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

PySpark'ta “huggingface” Veri Kaynağı aracılığıyla Hugging Face veri seti depolarına erişmek için pyspark_huggingface kullanabilirsiniz.

Hugging Face Spaces'te 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 göz atın.

“huggingface” Veri Kaynağını Etkinleştir

PySpark 4, özel kaynaklardan veri setleri kullanmaya olanak tanıyan 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ğına yönelik Veri Kaynağı 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:

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

Oku

“huggingface” Veri Kaynağı, pyarrow altyapısını kullanarak Arrow verilerini akış olarak okumak için Hugging Face'ten veri setlerini okumaya olanak tanır. Bu, Hugging Face'teki desteklenen biçimlerdeki tüm veri setleriyle, örneğin Parquet veri setleriyle uyumludur.

Örneğin stanfordnlp/imdb veri setinin nasıl yükleneceği aşağıdaki gibidir:

>>> 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 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üklenmesi gereken alt kümenin belirtilmesi gerekir.

“huggingface” 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:

>>> 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 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.

>>> 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:

>>> 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üklerken yapmak daha verimlidir. Gerçekten de 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'e iletilir.

Diğer tüm seçenekler, [datasets.load_dataset] (https://huggingface.co/docs/datasets/en/package_reference/loading_methods#datasets.load_dataset) işlevine 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:

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

“huggingface” Veri Kaynağı ile bir PySpark Dataframe'ini Hugging Face'e yazabilirsiniz. Parquet dosyalarını dağıtık bir şekilde paralel olarak yükler ve dosyaların tamamı yüklendikten sonra yalnızca dosyaları onaylar. İşleyişi şu şekildedir:

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

Bu fonksiyonu kullanarak 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 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:

>>> 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”: veri seti zaten varsa ü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 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:

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