Spark
Spark
Spark는 분산 환경에서 실시간 대규모 데이터 처리를 가능하게 해요. pyspark_huggingface로 PySpark에서 "huggingface" Data Source를 통해 Hugging Face 데이터셋 저장소에 접근할 수 있습니다.
출처: 문서
본문
Spark는 분산 환경에서 실시간 대규모 데이터 처리를 가능하게 합니다.
pyspark_huggingface를 사용해 PySpark에서 "huggingface" Data Source를 통해 Hugging Face 데이터셋 저장소에 접근할 수 있어요.
Spark Notebooks를 Hugging Face Spaces에서 사용해 PySpark와 pyspark_huggingface가 사전 설치된 Notebook을 써 보세요.
설정 (Set up)
설치 (Installation)
Hugging Face Datasets를 읽고 쓰려면 pyspark_huggingface 라이브러리를 설치해야 합니다:
pip install pyspark_huggingface
이것은 인증용 huggingface_hub, 데이터셋 읽고 쓰기용 pyarrow 같은 필요한 의존성도 설치합니다.
인증 (Authentication)
비공개/게이트된 데이터셋 저장소를 읽거나 내 데이터셋 저장소에 쓰려면 Hugging Face에 인증해야 합니다.
예를 들어 CLI를 사용할 수 있어요:
hf auth login
HF_TOKEN 환경 변수로 Hugging Face 토큰을 제공하거나, reader에 token 옵션을 전달하는 것도 가능합니다. 인증에 대한 자세한 내용은 이 가이드를 확인하세요.
"huggingface" Data Source 활성화 (Enable the "huggingface" Data Source)
PySpark 4는 커스텀 소스의 데이터셋을 사용할 수 있는 새 Data Source API를 도입했습니다.
pyspark_huggingface가 설치되면 PySpark가 자동으로 가져와 "huggingface" Data Source를 활성화합니다.
이 라이브러리는 PySpark 3.5, 3.4, 3.3용 "huggingface" Data Source의 Data Source API도 백포트합니다.
다만 이 경우 pyspark_huggingface를 명시적으로 가져와 백포트를 활성화하고 "huggingface" Data Source를 켜야 합니다:
>>> import pyspark_huggingface
huggingface datasource enabled for pyspark 3.x.x (backport from pyspark 4)
읽기 (Read)
"huggingface" Data Source를 사용하면 내부적으로 pyarrow로 Arrow 데이터를 스트리밍해 Hugging Face에서 데이터셋을 읽을 수 있어요. Parquet 데이터셋처럼 Hugging Face의 지원 형식에 있는 모든 데이터셋과 호환됩니다.
예를 들어 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" Data Source를 사용하고 .load()로 데이터셋(더 정확히는 "7M"이라는 이름의 config/하위 집합으로 7M 샘플을 담고 있음)을 불러옵니다. 그런 다음 언어별 대화 수를 계산하고 데이터셋을 필터링합니다.
게이트된 저장소에 접근하기 위해 로그인한 뒤 다음을 실행할 수 있어요:
>>> 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() 연산을 사용하는 이 코드를 실행합니다.
PySpark는 Data Source API에서 predicate push-down을 지원하지 않으므로 columns 옵션이 필요한 데이터만 불러오는 데 유용합니다.
특정 범위의 값만 가진 데이터만 불러오는 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의 컬럼도 독립적으로 로드할 수 있어 제외된 컬럼을 건너뛰고 불필요한 데이터 로드를 피할 수 있습니다.
옵션 (Options)
read..option()에 전달할 수 있는 사용 가능한 옵션 목록입니다:
config(string): 데이터셋 하위 집합/config 선택split(string): 데이터셋 스플릿 선택 (기본값 "train")token(string): 내 Hugging Face 토큰
config나 split을 지정하는 대신 로드할 파일을 수동으로 선택할 수 있습니다:
data_dir(string): 디렉터리 선택data_files(string): 하나 또는 여러 파일 선택, 예:"data/*.parquet"또는'["part1.parquet", "par2.parquet"]'
Parquet 데이터셋의 경우:
columns(string): 로드할 컬럼 부분집합 선택, 예:'["id"]'filters(string): 기준과 일치하지 않는 파일·행 그룹을 건너뜀, 예:'[("source", "=", "code_exercises")]'. 필터는 pyarrow.parquet.ParquetDataset에 전달됩니다.
그 외의 옵션은 datasets.load_dataset의 인자로 전달됩니다.
SQL 쿼리 실행 (Run SQL queries)
PySpark DataFrame이 준비되면 spark.sql로 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 옵션을 지정하는 것은 필수가 아니지만, 불필요한 데이터 로드를 피하고 쿼리를 더 빠르게 하는 데 유용합니다.
쓰기 (Write)
"huggingface" Data Source로 PySpark DataFrame을 Hugging Face에 쓸 수 있어요. Parquet 파일을 분산 방식으로 병렬 업로드하고, 모든 파일이 업로드된 후에만 커밋합니다. 다음과 같이 동작합니다:
>>> import pyspark_huggingface
>>> df.write.format("huggingface").save("username/dataset_name")
이 함수로 BAAI/Infinity-Instruct 데이터셋의 필터링된 버전을 Hugging Face에 다시 쓰는 방법을 보여줍니다.
먼저 데이터셋 저장소를 만들고(예: username/Infinity-Instruct-Chinese-Only, 원하면 private으로 설정 가능), 인증되었는지 확인하고 "huggingface" Data Source를 사용할 수 있는지 확인한 뒤 mode를 "overwrite"로 설정하고(기존 데이터셋을 확장하려면 "append"), .save()로 Hugging Face에 푸시합니다:
>>> df_chinese_only.write.format("huggingface").mode("overwrite").save("username/Infinity-Instruct-Chinese-Only")
모드 (Mode)
데이터셋을 Hugging Face에 푸시할 때 두 가지 모드를 사용할 수 있습니다:
- "overwrite": 이미 있으면 데이터셋 덮어쓰기
- "append": 기존 데이터셋에 데이터셋 추가
옵션 (Options)
write.option()에 전달할 수 있는 사용 가능한 옵션 목록입니다:
token(string): 내 Hugging Face 토큰
특히 subset과 split 같은 더 많은 옵션을 여기에 추가하는 기여를 환영합니다.
Storage Buckets
Storage Buckets에서 원시 데이터를 처리하고 실험한 뒤 AI 준비가 된 데이터를 Dataset 저장소에 게시하는 것이 일반적입니다.
buckets/ 접두사와 data_dir 또는 data_files 옵션을 사용해 Dataset 저장소와 같은 방식으로 Storage Buckets에 접근하세요:
>>> 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")
더 알아보기 (Learn more)
Spark(PySpark)로 pyspark_huggingface를 설치하면 .read.format("huggingface").load("repo")로 Hugging Face 데이터셋을 스트리밍 로드하고, config/split/columns/filters/token 옵션을 지원해요. 쓰기는 df.write.format("huggingface").mode("overwrite|append").save("username/repo")로 분산 Parquet 업로드가 가능하며, buckets/ 접두사로 Storage Buckets도 다룰 수 있습니다. 게이트된 저장소는 먼저 로그인해야 해요.