Spark

Spark обеспечивает обработку данных в реальном времени в больших масштабах в распределённой среде.

Вы можете использовать pyspark_huggingface для доступа к репозиториям датасетов Hugging Face в PySpark через источник данных “huggingface”.

Попробуйте Spark Notebooks на Hugging Face Spaces, чтобы получить Notebooks с предварительно установленными PySpark и pyspark_huggingface.

Настройка

Установка

Чтобы читать и записывать данные в Hugging Face Datasets, необходимо установить библиотеку pyspark_huggingface:

pip install pyspark_huggingface

Это также установит необходимые зависимости, такие как huggingface_hub для аутентификации и pyarrow для чтения и записи наборов данных.

Аутентификация

Для чтения частных/защищённых репозиториев датасетов или записи в ваши репозитории датасетов необходимо пройти аутентификацию в Hugging Face.

Вы можете использовать CLI, например:

hf auth login

Также можно передать токен Hugging Face с помощью переменной окружения HF_TOKEN или опции token при создании читателя. Подробнее об аутентификации — в этом руководстве.

Включите источник данных “huggingface”

PySpark 4 поставляется с новым Data Source API, который позволяет использовать наборы данных из пользовательских источников. Если pyspark_huggingface установлен, PySpark автоматически импортирует его и активирует Data Source «huggingface».

Библиотека также обеспечивает обратную совместимость (backport) API источника данных для источника данных “huggingface” в PySpark 3.5, 3.4 и 3.3. Однако в этом случае pyspark_huggingface необходимо импортировать явно, чтобы активировать backport и включить источник данных “huggingface”:

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

Читать

Источник данных «huggingface» позволяет читать наборы данных с Hugging Face, используя pyarrow для потоковой передачи данных в формате Arrow. Это совместимо со всеми наборами данных в поддерживаемых форматах на 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”, а .load() — для загрузки набора данных (точнее, конфигурации или подмножества с именем “7M”, содержащего 7 млн образцов). Затем мы вычисляем количество диалогов для каждого языка и фильтруем набор данных.

После входа в систему для доступа к защищённому репозиторию мы можем выполнить:

>>> 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 не поддерживает предикатный push-down с использованием Data Source API. Также существует опция 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 содержит метаданные на уровне файла и группы строк, что позволяет пропускать целые части датасета, не содержащие образцы, удовлетворяющие заданным критериям. Столбцы в Parquet также можно загружать независимо, что позволяет пропускать исключаемые столбцы и избегать загрузки ненужных данных.

Параметры

Вот список доступных опций, которые можно передать в read..option():

• config (строка): выбрать подмножество/конфигурацию набора данных
• split (string): выбрать разделение набора данных (по умолчанию — «train»)
• token (строка): ваш токен Hugging Face

Вместо указания конфигурации или сплита вы можете вручную выбрать файлы для загрузки:

• data_dir (строка): выберите каталог
• data_files (строка): выберите один или несколько файлов, например "data/*.parquet" или '["part1.parquet", "par2.parquet"]'

Для наборов данных в формате Parquet:

• columns (string): выбор подмножества столбцов для загрузки, например '["id"]'
• filters (string): для пропуска файлов и групп строк, не соответствующих заданным условиям, например '[("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 готов, вы можете выполнять SQL-запросы с помощью spark.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 не является обязательным, но полезно, чтобы избежать загрузки ненужных данных и ускорить выполнение запроса.

Запись

Вы можете записать PySpark Dataframe в Hugging Face с помощью источника данных “huggingface”. Он загружает файлы Parquet параллельно в распределённом режиме и фиксирует файлы только после того, как все они будут загружены. Это работает следующим образом:

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

Вот как мы можем использовать эту функцию, чтобы записать отфильтрованную версию набора данных BAAI/Infinity-Instruct обратно в Hugging Face.

Сначала вам нужно создать репозиторий датасета, например username/Infinity-Instruct-Chinese-Only (при желании вы можете сделать его приватным). Затем убедитесь, что вы аутентифицированы и можете использовать источник данных “huggingface”, установите mode в значение “overwrite” (или “append”, если вы хотите расширить существующий датасет), и отправьте данные на Hugging Face с помощью .save():

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

Режим

При загрузке набора данных на Hugging Face доступны два режима:

• “overwrite”: перезаписать набор данных, если он уже существует
• “append”: добавить набор данных к существующему

Параметры

Вот список доступных опций, которые можно передать в write.option():

• token (строка): ваш токен Hugging Face

Мы приветствуем предложения по добавлению дополнительных вариантов, в частности subset и split.

Бuckets для хранения

Обычно сырые данные обрабатывают в Storage Buckets и проводят эксперименты там, прежде чем публиковать данные, готовые для ИИ, в репозиториях датасетов.

Доступ к хранилищам (Storage Buckets) осуществляется так же, как к репозиториям Dataset, но с префиксом 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")