Hub documentation
Spark
Spark
Spark cho phép xử lý dữ liệu quy mô lớn theo thời gian thực trong môi trường phân tán.
Bạn có thể sử dụng pyspark_huggingface để truy cập các kho dữ liệu Hugging Face trong PySpark thông qua Data Source “huggingface”.
Thử Spark Notebooks trên Hugging Face Spaces để có các Notebooks đã cài sẵn PySpark và pyspark_huggingface.
Cài đặt
Cài đặt
Để có thể đọc và ghi vào Hugging Face Datasets, bạn cần cài đặt thư viện pyspark_huggingface:
pip install pyspark_huggingfaceĐiều này cũng sẽ cài đặt các phụ thuộc cần thiết như huggingface_hub cho xác thực, và pyarrow để đọc và ghi các bộ dữ liệu.
Xác thực
Bạn cần xác thực với Hugging Face để đọc các kho dữ liệu riêng tư/có kiểm soát hoặc để ghi vào các kho dữ liệu của bạn.
Bạn có thể sử dụng CLI, ví dụ:
hf auth loginBạn cũng có thể cung cấp token Hugging Face của mình bằng biến môi trường HF_TOKEN hoặc bằng cách truyền tùy chọn token cho reader.
Để biết thêm chi tiết về xác thực, hãy xem hướng dẫn này.
Bật Nguồn dữ liệu “huggingface”
PySpark 4 đi kèm với Data Source API mới, cho phép sử dụng các dataset từ các nguồn tùy chỉnh.
Nếu pyspark_huggingface được cài đặt, PySpark sẽ tự động import nó và kích hoạt Data Source “huggingface”.
Thư viện cũng backport API Data Source cho Data Source “huggingface” của PySpark 3.5, 3.4 và 3.3.
Tuy nhiên, trong trường hợp này, pyspark_huggingface cần được import một cách tường minh để kích hoạt backport và bật Data Source “huggingface”:
>>> import pyspark_huggingface
huggingface datasource enabled for pyspark 3.x.x (backport from pyspark 4)Đọc
Nguồn dữ liệu “huggingface” cho phép đọc các bộ dữ liệu từ Hugging Face, sử dụng pyarrow bên trong để truyền trực tiếp dữ liệu Arrow.
Điều này tương thích với tất cả các bộ dữ liệu ở định dạng được hỗ trợ trên Hugging Face, chẳng hạn như các bộ dữ liệu Parquet.
Ví dụ, đây là cách tải bộ dữ liệu stanfordnlp/imdb:
>>> import pyspark_huggingface
>>> from pyspark.sql import SparkSession
>>> spark = SparkSession.builder.appName("demo").getOrCreate()
>>> df = spark.read.format("huggingface").load("stanfordnlp/imdb")Dưới đây là một ví dụ khác với bộ dữ liệu BAAI/Infinity-Instruct. Đây là một kho lưu trữ có kiểm soát, người dùng phải chấp nhận các điều khoản sử dụng trước khi truy cập. Nó cũng có nhiều tập con, cụ thể là “3M” và “7M”. Vì vậy, chúng ta cần chỉ định tập con nào để tải.
Chúng ta sử dụng hàm .format() để dùng Data Source “huggingface”, và .load() để tải dataset (chính xác hơn là config hoặc subset có tên “7M” chứa 7M mẫu). Sau đó, chúng ta tính số lượng hội thoại cho từng ngôn ngữ và lọc dataset.
Sau khi đăng nhập để truy cập kho lưu trữ có kiểm soát, chúng ta có thể chạy:
>>> 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|
+---+----------------------------+-----+----------+--------------------+Điều này tải bộ dữ liệu theo kiểu streaming, và DataFrame đầu ra có một partition cho mỗi tệp dữ liệu trong bộ dữ liệu để cho phép xử lý phân tán hiệu quả.
Để tính số lượng hội thoại cho từng ngôn ngữ, chúng ta chạy đoạn mã này sử dụng tùy chọn columns và thao tác groupBy().
Tùy chọn columns hữu ích để chỉ tải dữ liệu cần thiết, vì PySpark không hỗ trợ predicate push-down với Data Source API.
Cũng có tùy chọn filters để chỉ tải dữ liệu có giá trị nằm trong một khoảng nhất định.
>>> 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|
+----------+-------+Để lọc bộ dữ liệu và chỉ giữ lại các cuộc đối thoại bằng tiếng Trung:
>>> 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|
+---+----------------------------+-----+----------+----------+Bạn cũng có thể áp dụng bộ lọc hoặc xóa các cột trên DataFrame đã tải, nhưng hiệu quả hơn là thực hiện điều này trong quá trình tải, đặc biệt là với các tập dữ liệu Parquet. Thật vậy, Parquet chứa metadata ở cấp độ tệp và nhóm hàng (row group), cho phép bỏ qua toàn bộ các phần của tập dữ liệu không chứa các mẫu thỏa mãn tiêu chí. Các cột trong Parquet cũng có thể được tải độc lập, cho phép bỏ qua các cột bị loại trừ và tránh việc tải dữ liệu không cần thiết.
Tùy chọn
Dưới đây là danh sách các tùy chọn có sẵn mà bạn có thể truyền vào read..option():
config(chuỗi): chọn một tập con/cấu hình của bộ dữ liệusplit(chuỗi): chọn một phần chia của bộ dữ liệu (mặc định là “train”)token(chuỗi): token Hugging Face của bạn
Thay vì chỉ định một config hoặc split, bạn có thể tự động chọn các tệp cần tải:
data_dir(string): chọn một thư mụcdata_files(string): chọn một hoặc nhiều tệp, ví dụ"data/*.parquet"hoặc'["part1.parquet", "par2.parquet"]'
Đối với các bộ dữ liệu Parquet:
columns(string): chọn một tập con các cột để tải, ví dụ'["id"]'filters(string): để bỏ qua các tệp và nhóm dòng không khớp với một tiêu chí, ví dụ'[("source", "=", "code_exercises")]'. Bộ lọc được truyền đến pyarrow.parquet.ParquetDataset.
Mọi tùy chọn khác được truyền như một tham số cho [datasets.load_dataset] (https://huggingface.co/docs/datasets/en/package_reference/loading_methods#datasets.load_dataset)
Chạy truy vấn SQL
Khi bạn đã chuẩn bị xong Dataframe PySpark, bạn có thể chạy truy vấn SQL bằng 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|
+--------------------+-------+Một lần nữa, việc chỉ định tùy chọn columns không bắt buộc, nhưng hữu ích để tránh nạp dữ liệu không cần thiết và giúp truy vấn nhanh hơn.
Viết
Bạn có thể ghi một PySpark Dataframe vào Hugging Face bằng Data Source “huggingface”. Nó tải các tệp Parquet lên song song theo cách phân tán, và chỉ cam kết các tệp khi tất cả đã được tải lên. Nó hoạt động như sau:
>>> import pyspark_huggingface
>>> df.write.format("huggingface").save("username/dataset_name")Dưới đây là cách chúng ta có thể sử dụng hàm này để ghi phiên bản đã lọc của bộ dữ liệu BAAI/Infinity-Instruct trở lại Hugging Face.
Trước tiên, bạn cần tạo một kho dữ liệu, ví dụ username/Infinity-Instruct-Chinese-Only (bạn có thể đặt nó ở chế độ riêng tư nếu muốn).
Sau đó, đảm bảo bạn đã xác thực và có thể sử dụng Nguồn dữ liệu “huggingface”, đặt mode thành “overwrite” (hoặc “append” nếu bạn muốn mở rộng một bộ dữ liệu hiện có), và đẩy lên Hugging Face bằng .save():
>>> df_chinese_only.write.format("huggingface").mode("overwrite").save("username/Infinity-Instruct-Chinese-Only")
Chế độ
Có hai chế độ có sẵn khi đẩy một tập dữ liệu lên Hugging Face:
- “overwrite”: ghi đè lên bộ dữ liệu nếu nó đã tồn tại
- “append”: thêm bộ dữ liệu vào một bộ dữ liệu hiện có
Tùy chọn
Dưới đây là danh sách các tùy chọn có sẵn mà bạn có thể truyền vào write.option():
token(chuỗi): token Hugging Face của bạn
Chào mừng đóng góp để thêm nhiều tùy chọn hơn tại đây, đặc biệt là subset và split.
Thùng chứa lưu trữ
Việc xử lý dữ liệu thô trong Storage Buckets và thử nghiệm tại đó trước khi công bố dữ liệu sẵn sàng cho AI trong các kho Dataset là rất phổ biến.
Truy cập Storage Buckets theo cách tương tự như các kho Dataset nhưng với tiền tố buckets/ và các tùy chọn data_dir hoặc 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")