⭐ PySpark 和 🤗 Hugging Face Parquet 檔案

簡介

歡迎閱讀本指南，瞭解如何使用 PySpark 載入和處理來自 Hugging Face 資料集的 Parquet 檔案！我們將引導您設定 Spark 會話、載入 Parquet 檔案並執行基本資料操作，所有這些都以葡萄酒評論資料集為例。讓我們開始吧！

---

目錄

1. 設定
2. 顯示資料集
3. 載入 Hugging Face Parquet 檔案
4. 探索資料
5. 資料轉換
6. 結論

---

1. 設定

開始之前，讓我們先設定環境。首先，我們將安裝必要的庫並啟動一個 Spark 會話。

pip install pyspark

接下來，我們將匯入所需的依賴項並初始化 Spark 會話。Spark 會話是您使用 Spark 處理 DataFrame 的入口點。

from pyspark import SparkFiles
from pyspark import SparkContext
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("WineReviews").getOrCreate()

---

2. 顯示資料集

我們可以直接從 Hugging Face 顯示一個互動式檢視器，以更好地理解我們的資料集。此步驟是可選的，但強烈推薦！

from IPython.display import HTML

dataset = "james-burton/wine_reviews"

iframe_html = """
<iframe src="https://huggingface.co/datasets/{dataset}/embed/viewer" width="80%" height="560px"></iframe>
""".format(dataset=dataset)
display(HTML(iframe_html))

葡萄酒評論資料集包含專業葡萄酒評論家和愛好者對葡萄酒的評論，詳細介紹了各種葡萄酒特徵，如品種、評分、價格和產地。

---

3. 載入 Hugging Face Parquet 檔案

現在，讓我們透過從 Hugging Face API 獲取 Parquet 檔案 URL 並將它們新增到 Spark 上下文中來載入資料集。

import requests

HUGGING_FACE_PARQUET_API = "https://huggingface.co/api/datasets/{dataset}/parquet"
r = requests.get(HUGGING_FACE_PARQUET_API.format(dataset=dataset))

train_parquet_files = r.json()['default']['train']

for url in train_parquet_files:
  spark.sparkContext.addFile(url)

df = spark.read.parquet(SparkFiles.getRootDirectory() + "/*.parquet")

---

4. 探索資料

將資料載入到 DataFrame 後，我們可以探索其結構和內容。

# Shape of the dataset
print(f"Shape of the dataset: {df.count()}, {len(df.columns)}")

# Displaying first 10 rows
df.show(n=10)

# Getting a statistical summary of the data
df.describe().show()

# Print the schema of the DataFrame
df.printSchema()

---

5. 資料轉換

讓我們對資料集進行一些基本轉換。

# Display all values of a particular column
df.select('country').show()

# Select multiple columns
df.select(['country','province']).show()

# Display data types of columns
df.dtypes

# Create a subset of the dataset
df1 = df.limit(5)
df1.show()

新增新列

我們將新增一個新列，將 `country` 和 `province` 的值用連字元連線起來。

from pyspark.sql.functions import concat, lit

df1 = df.withColumn("location", concat(df['country'], lit('-'), df['province']))
df1.show()

分組資料

按國家/地區對資料進行分組，並計算每個國家/地區的記錄數，然後按降序對結果進行排序。

df.groupBy('country').count().orderBy('count', ascending=False).show()

執行 SQL 查詢

您也可以使用 SQL 查詢實現相同的結果。

df.createOrReplaceTempView("wine_reviews_table")
spark.sql("SHOW TABLES;").show()

result_df = spark.sql("SELECT country, count(*) as count from wine_reviews_table GROUP BY country ORDER BY count DESC")
result_df.show()

處理缺失值

讓我們檢查並處理資料集中的缺失值。

from pyspark.sql.functions import col, isnan, when, count

null_df_counts = df.select([count(when(col(c).contains('None') | \
                            col(c).contains('NULL') | \
                            (col(c) == '') | \
                            col(c).isNull() | \
                            isnan(c), c
                           )).alias(c)
                    for c in df.columns])

null_df_counts.show()

# Removing rows with any null values
df_clean = df.dropna()
df_clean.show()

---

您可以在此處檢視包含程式碼的筆記本。

6. 結論

在本教程中，我們介紹了：

• 初始化 Spark 會話。
• 將 Hugging Face 的 Parquet 檔案載入到 PySpark DataFrame 中。
• 探索資料結構和內容。
• 執行資料轉換。
• 執行 SQL 查詢。
• 處理缺失值。

這些步驟為使用 PySpark 進行更高階的資料分析和轉換奠定了基礎。

🤗 祝您資料處理愉快！