Spark

Spark permite el procesamiento de datos a gran escala en tiempo real en un entorno distribuido.

Puede usar pyspark_huggingface para acceder a los repositorios de conjuntos de datos de Hugging Face en PySpark a través de la fuente de datos “huggingface”.

Prueba Spark Notebooks en Hugging Face Spaces para obtener Notebooks con PySpark y pyspark_huggingface preinstalados.

Configuración

Instalación

Para poder leer y escribir en Hugging Face Datasets, necesitas instalar la biblioteca pyspark_huggingface:

pip install pyspark_huggingface

Esto también instalará las dependencias necesarias, como huggingface_hub para la autenticación y pyarrow para leer y escribir conjuntos de datos.

Autenticación

Debes autenticarte en Hugging Face para leer repositorios de datos privados o con acceso restringido, o para escribir en tus repositorios de datos.

Puedes usar la CLI, por ejemplo:

hf auth login

También es posible proporcionar tu token de Hugging Face con la variable de entorno HF_TOKEN o pasando la opción token al lector. Para más detalles sobre la autenticación, consulta esta guía.

Activar la fuente de datos “huggingface”

PySpark 4 incorporó una nueva API de origen de datos que permite utilizar conjuntos de datos de fuentes personalizadas. Si pyspark_huggingface está instalado, PySpark lo importa automáticamente y habilita el origen de datos “huggingface”.

La biblioteca también retroporta la API de Data Source para el Data Source “huggingface” para PySpark 3.5, 3.4 y 3.3. Sin embargo, en este caso pyspark_huggingface debe importarse explícitamente para activar la retroportación y habilitar el Data Source “huggingface”:

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

Leer

La fuente de datos “huggingface” permite leer conjuntos de datos de Hugging Face, utilizando pyarrow internamente para transmitir datos Arrow. Esto es compatible con todos los conjuntos de datos en formato compatible en Hugging Face, como los conjuntos de datos Parquet.

Por ejemplo, así es como se carga el conjunto de datos stanfordnlp/imdb:

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

Aquí hay otro ejemplo con el conjunto de datos BAAI/Infinity-Instruct. Es un repositorio con acceso restringido, los usuarios deben aceptar los términos de uso antes de acceder a él. También tiene varios subconjuntos, a saber, “3M” y “7M”. Por lo tanto, necesitamos especificar cuál de ellos cargar.

Usamos la función .format() para usar la fuente de datos “huggingface”, y .load() para cargar el conjunto de datos (más precisamente, la configuración o subconjunto llamado “7M” que contiene 7M muestras). Luego, calculamos el número de diálogos por idioma y filtramos el conjunto de datos.

Después de iniciar sesión para acceder al repositorio restringido, podemos ejecutar:

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

Esto carga el conjunto de datos de forma de streaming, y el DataFrame de salida tiene una partición por archivo de datos en el conjunto de datos para permitir un procesamiento distribuido eficiente.

Para calcular el número de diálogos por idioma, ejecutamos este código que utiliza la opción columns y una operación groupBy(). La opción columns es útil para cargar solo los datos que necesitamos, ya que PySpark no habilita el empuje de predicados con la API de Data Source. También existe una opción filters para cargar solo datos con valores dentro de un cierto rango.

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

Para filtrar el conjunto de datos y conservar solo los diálogos en chino:

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

También es posible aplicar filtros o eliminar columnas en el DataFrame cargado, pero es más eficiente hacerlo durante la carga, especialmente en conjuntos de datos Parquet. De hecho, Parquet contiene metadatos a nivel de archivo y grupo de filas, lo que permite omitir partes enteras del conjunto de datos que no contienen muestras que cumplan los criterios. Las columnas en Parquet también se pueden cargar de forma independiente, lo que permite omitir las columnas excluidas y evitar la carga de datos innecesarios.

Opciones

Esta es la lista de opciones disponibles que puedes pasar a read..option():

• config (string): selecciona un subconjunto/configuración del conjunto de datos
• split (string): selecciona una división del conjunto de datos (el valor predeterminado es “train”)
• token (string): tu token de Hugging Face

En lugar de especificar una configuración o una división, puede seleccionar manualmente qué archivos cargar:

• data_dir (string): selecciona un directorio
• data_files (string): selecciona uno o varios archivos, p. ej. "data/*.parquet" o '["part1.parquet", "par2.parquet"]'

Para conjuntos de datos Parquet:

• columns (string): selecciona un subconjunto de columnas para cargar, p. ej. '["id"]'
• filters (string): para omitir archivos y grupos de filas que no cumplan un criterio, p. ej. '[("source", "=", "code_exercises")]'. Los filtros se pasan a pyarrow.parquet.ParquetDataset.

Cualquier otra opción se pasa como argumento a [datasets.load_dataset] (https://huggingface.co/docs/datasets/en/package_reference/loading_methods#datasets.load_dataset)

Ejecutar consultas SQL

Una vez que tengas tu Dataframe de PySpark listo, puedes ejecutar consultas SQL usando 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|
+--------------------+-------+

Nuevamente, especificar la opción columns no es necesario, pero resulta útil para evitar cargar datos innecesarios y hacer que la consulta sea más rápida.

Escribir

Puedes escribir un Dataframe de PySpark en Hugging Face con la fuente de datos “huggingface”. Sube archivos Parquet en paralelo de forma distribuida y solo confirma los archivos una vez que todos se han subido. Funciona de la siguiente manera:

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

A continuación se muestra cómo podemos usar esta función para escribir la versión filtrada del conjunto de datos BAAI/Infinity-Instruct de nuevo en Hugging Face.

Primero necesitas crear un repositorio de datos, por ejemplo username/Infinity-Instruct-Chinese-Only (puedes configurarlo como privado si lo deseas). A continuación, asegúrate de que estás autenticado y de que puedes usar la fuente de datos “huggingface”, establece mode en “overwrite” (o en “append” si quieres ampliar un conjunto de datos existente) y sube a Hugging Face con .save():

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

Modo

Hay dos modos disponibles al subir un conjunto de datos a Hugging Face:

• “overwrite”: sobrescribir el conjunto de datos si ya existe
• “append”: añade el conjunto de datos a un conjunto de datos existente

Opciones

Esta es la lista de opciones disponibles que puedes pasar a write.option():

• token (string): tu token de Hugging Face

Se aceptan contribuciones para añadir más opciones aquí, en particular subset y split.

Baldes de almacenamiento

Es habitual procesar datos crudos en Storage Buckets y experimentar allí antes de publicar datos listos para IA en repositorios de Dataset.

Acceda a los buckets de almacenamiento de la misma manera que a los repositorios de Dataset, pero con el prefijo buckets/ y con las opciones data_dir o 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")