PySpark 在處理大規模資料時，效能最佳化對於深度學習訓練至關重要。透過調整資料分割槽策略，可以有效利用叢集資源，平行處理資料。控制洗牌分割槽數量能減少資料傳輸成本，提升運算效率。此外，使用 Parquet 格式儲存資料，結合快取和廣播等技巧，能進一步提升讀取和運算速度。啟用 AQE 則允許 Spark 動態調整查詢計畫，最佳化執行效能。
最佳化PySpark資料處理效能以利深度學習
在進行大規模資料處理時，PySpark的效能最佳化至關重要。適當的組態和技巧可以大幅減少處理時間並提升整體運算效率。本章節將探討如何利用PySpark的最佳化技術來提升資料處理效能，以支援深度學習專案的需求。
資料分割槽的調整
在PySpark中，資料的分割槽（partition）對效能有著直接的影響。當資料僅存在於一個分割槽時，Spark只能依序處理資料，無法充分利用叢集中的多個核心。透過repartition()函式，可以將資料重新分割槽，以實作平行處理。
程式碼範例：重新分割槽
repartitioned_df = initial_df.repartition(10)
data_processor.print_partition_info(repartitioned_df)
內容解密：
repartition(10)將原始DataFrame重新分割槽為10個分割槽。data_processor.print_partition_info(repartitioned_df)用於列印重新分割槽後的DataFrame分割槽數量。- 透過增加分割槽數量，可以提高平行度，加快資料處理速度。
調整分割槽數量
在某些情況下，可能需要減少分割槽數量以降低管理成本。coalesce()函式可以在減少分割槽的同時，盡量減少資料的洗牌（shuffle）。
程式碼範例：合併分割槽
coalesced_df = repartitioned_df.coalesce(5)
data_processor.print_partition_info(coalesced_df)
內容解密：
coalesce(5)將分割槽數量從10減少到5。- 減少分割槽數量的同時，也減少了平行度，因此需要根據實際情況進行權衡。
設定洗牌分割槽數量
在進行需要洗牌的操作（如join、groupBy）時，spark.sql.shuffle.partitions組態決定了洗牌後的分割槽數量。適當調整此引數可以提升效能。
程式碼範例：設定洗牌分割槽數量
spark.conf.set("spark.sql.shuffle.partitions", "20")
內容解密：
spark.sql.shuffle.partitions預設值為200，但在某些場景下，設定為20可能更為合適。- 適當的洗牌分割槽數量可以減少不必要的資料傳輸，提升運算效率。
儲存資料為Parquet格式
將DataFrame儲存為Parquet格式不僅可以節省儲存空間，還能因其最佳化的讀取操作而提升效能。
程式碼範例：儲存為Parquet
repartitioned_df.write.parquet("/home/ubuntu/airflow/dags/tesla_stock.parquet")
內容解密：
write.parquet()將DataFrame儲存為Parquet檔案。- Parquet格式支援壓縮和欄位儲存，能夠加快讀取速度並節省空間。
快取DataFrame
對於需要多次使用的DataFrame，快取（cache）可以避免重複計算，從而提升效能。
程式碼範例：快取DataFrame
repartitioned_df.cache()
內容解密：
cache()將DataFrame快取在記憶體中。- 快取適合於多次使用的DataFrame，能夠顯著提升後續操作的效率。
廣播小資料表
在進行join操作時，如果其中一個DataFrame較小，可以使用廣播（broadcast）來避免洗牌，從而提升效能。
程式碼範例：廣播join
from pyspark.sql.functions import broadcast
small_df = spark.createDataFrame([
("2023-08-25", 5.5),
("2023-08-26", 6.0),
], ["Date", "AdditionalData"])
joined_df = repartitioned_df.join(broadcast(small_df), on="Date", how="left")
內容解密：
broadcast(small_df)將小資料表廣播到所有工作節點。- 使用廣播join可以避免大資料表的洗牌，大幅提升join操作的效率。
啟用自適應查詢執行（AQE）
自適應查詢執行（Adaptive Query Execution, AQE）能夠根據執行時的統計資訊動態調整查詢計畫，從而最佳化效能。
程式碼範例：啟用AQE
spark.conf.set("spark.sql.adaptive.enabled", "true")
內容解密：
spark.sql.adaptive.enabled啟用AQE功能。- AQE特別適用於使用DataFrame和Spark SQL API的操作，能夠動態最佳化join策略和洗牌分割槽數量。
綜合應用範例
以下是一個綜合了上述最佳化技巧的Tesla股票價格資料處理指令碼：
程式碼範例：綜合應用
import boto3
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast
class DataProcessor:
def __init__(self, spark_session):
self.spark = spark_session
def load_data(self, file_path: str):
try:
df = self.spark.read.csv(file_path, header=True, inferSchema=True)
return df
except Exception as e:
print(f"Error loading data: {str(e)}")
return None
def print_first_n_rows(self, df, n=10):
print(f"First {n} rows of the DataFrame:")
df.show(n)
def print_partition_info(self, df):
num_partitions = df.rdd.getNumPartitions()
print(f"Number of partitions: {num_partitions}")
def main(s3_bucket_name: str, s3_file_key: str, local_file_path: str):
spark = SparkSession.builder \
.appName("StockPriceOptimization") \
.config("spark.sql.adaptive.enabled", "true") \
.config("spark.sql.shuffle.partitions", "20") \
.getOrCreate()
# 資料載入與處理邏輯...
內容解密：
- 上述程式碼建立了一個
DataProcessor類別來封裝資料處理的邏輯。 - 在
main函式中，建立了SparkSession並啟用了AQE和設定了洗牌分割槽數量。 - 這樣的設計能夠靈活地整合各種最佳化技巧，以滿足不同的資料處理需求。
資料準備與 PySpark 在深度學習中的應用
在探討特斯拉股票資料集並展示如何提升 PySpark 的平行處理能力後，我們現在將焦點轉向資料準備在深度學習框架中的應用，首先聚焦於 PyTorch。具體來說，我們將探討如何使用 PySpark 進行資料預處理、特徵工程和資料格式化。
匯入必要的模組
首先，我們需要匯入必要的模組和函式庫來執行資料準備任務並與 S3 互動。
import logging
from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.sql import DataFrame
import numpy as np
import torch
from torch.utils.data import DataLoader, TensorDataset
import boto3
內容解密：
- logging：用於在程式執行過程中記錄訊息、警告和錯誤。
- SparkSession：用於建立 SparkSession 物件，以讀取 CSV 資料並建立 DataFrame 物件。
- VectorAssembler 和 StandardScaler：用於特徵工程，將原始特徵欄位彙編成單一特徵向量欄位，並對輸入特徵進行標準化處理。
- DataFrame：用於表示和操作表格格式的資料。
- numpy：用於將 Spark DataFrame 欄位轉換為 NumPy 陣列以進行進一步處理。
- torch 和 DataLoader, TensorDataset：用於建立 PyTorch DataLoader 以進行批次處理，處理張量並建立資料載入器以便於模型訓練。
- boto3：用於與 AWS S3 互動，下載指定 S3 儲存桶中的檔案到本地檔案路徑。
設定日誌紀錄
接下來，我們設定 Python 指令碼的日誌紀錄。
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
內容解密：
logging.basicConfig(level=logging.INFO)：設定日誌系統，僅記錄嚴重性等級為 INFO 及以上的訊息。logger = logging.getLogger(__name__)：建立一個名為logger的日誌物件，根據目前模組名稱命名，可用於記錄不同程式碼部分的日誌訊息。
載入資料
定義一個名為 load_data 的函式，使用 SparkSession 從 CSV 檔案載入特斯拉股票價格資料。
def load_data(file_path: str) -> DataFrame:
"""
使用 SparkSession 從 CSV 檔案載入股票價格資料。
"""
spark = (SparkSession.builder
.appName("StockPricePrediction")
.getOrCreate())
df = spark.read.csv(
file_path,
header=True,
inferSchema=True
)
return df
內容解密：
def load_data(file_path: str) -> DataFrame:：定義load_data函式，接受一個file_path引數，回傳一個 DataFrame 物件。spark = SparkSession.builder.appName("StockPricePrediction").getOrCreate()：建立一個 SparkSession 物件，並設定應用程式名稱為 “StockPricePrediction”。df = spark.read.csv(file_path, header=True, inferSchema=True)：從指定的 CSV 檔案讀取資料，建立 DataFrame 物件。
使用 PySpark 進行資料準備
在接下來的章節中，我們將探討如何使用 PySpark 進行資料預處理、特徵工程和資料格式化，以準備用於深度學習模型訓練的資料。
# 從 S3 下載檔案到本地
copy_file_from_s3(s3_bucket_name, s3_file_key, local_file_path)
# 建立 DataProcessor 物件
data_processor = DataProcessor(spark)
# 載入資料
df = data_processor.load_data(local_file_path)
if df is not None:
# 列印分割區資訊
data_processor.print_partition_info(df)
# 重新分割 DataFrame
repartitioned_df = df.repartition(10)
data_processor.print_partition_info(repartitioned_df)
# 合併分割區
coalesced_df = repartitioned_df.coalesce(5)
data_processor.print_partition_info(coalesced_df)
# 範例：廣播變數 join
small_df = spark.createDataFrame([
("2023-08-25", 5.5),
("2023-08-26", 6.0),
], ["Date", "AdditionalData"])
joined_df = repartitioned_df.join(broadcast(small_df), on="Date", how="left")
# 將 DataFrame 儲存為 Parquet 檔案
repartitioned_df.write.parquet("/home/ubuntu/airflow/dags/tesla_stock.parquet")
# 列印前幾行資料
data_processor.print_first_n_rows(repartitioned_df)
else:
print("載入資料失敗，結束程式。")
內容解密：
copy_file_from_s3：從 S3 下載檔案到本地檔案路徑。DataProcessor：自定義的類別，用於處理資料相關的操作，如載入資料、列印分割區資訊等。repartition和coalesce：用於調整 DataFrame 的分割區數量，以最佳化平行處理效率。broadcast join：使用廣播變數進行 join 操作，以提高效能。write.parquet：將 DataFrame 儲存為 Parquet 檔案格式。
使用 PySpark 進行資料準備以支援深度學習模型訓練
在建構深度學習模型之前，資料準備是一個至關重要的步驟。PySpark 提供了一個強大的工具集，能夠有效地處理和分析大規模的資料集。本篇文章將介紹如何使用 PySpark 進行資料載入、預處理和特徵工程，以為後續的深度學習模型訓練做好準備。
建立 SparkSession 並載入 CSV 資料
首先，我們需要建立一個 SparkSession，它是使用 PySpark 的入口點。接著，我們定義一個函式 load_data，該函式負責從指定的 CSV 檔案路徑載入資料。
載入資料函式實作
from pyspark.sql import SparkSession, DataFrame
def load_data(spark: SparkSession, file_path: str) -> DataFrame:
"""
載入 CSV 檔案至 DataFrame。
Parameters:
spark (SparkSession): SparkSession 物件。
file_path (str): CSV 檔案的路徑。
Returns:
DataFrame: 包含 CSV 資料的 DataFrame。
"""
df = spark.read.csv(file_path, header=True, inferSchema=True)
return df
載入資料函式解析：
spark.read.csv(file_path, header=True, inferSchema=True)：使用 SparkSession 的read.csv方法讀取 CSV 檔案。其中，header=True表示第一行是欄位名稱，inferSchema=True表示自動推斷欄位的資料型別。return df：傳回包含 CSV 資料的 DataFrame。
資料預處理
接下來，我們定義一個 preprocess_data 函式，用於對載入的資料進行預處理，包括特徵向量彙編、標準化等步驟。
資料預處理函式實作
from pyspark.ml.feature import VectorAssembler, StandardScaler
def preprocess_data(df: DataFrame) -> DataFrame:
"""
對資料進行預處理，包括特徵向量彙編和標準化。
Parameters:
df (DataFrame): 輸入的 DataFrame。
Returns:
DataFrame: 預處理後的 DataFrame。
"""
assembler = VectorAssembler(inputCols=['Open', 'High', 'Low', 'Volume'], outputCol='features')
df = assembler.transform(df)
logger.info("First 5 observations after assembling features:")
df.show(5, truncate=False)
df = perform_feature_engineering(df)
logger.info("First 5 observations after feature engineering:")
df.show(5, truncate=False)
scaler = StandardScaler(inputCol="features", outputCol="scaled_features", withStd=True, withMean=True)
scaler_model = scaler.fit(df)
df = scaler_model.transform(df)
df = df.select('scaled_features', 'Close')
return df
資料預處理函式解析：
VectorAssembler：將多個欄位彙編成一個向量欄位。StandardScaler：對特徵進行標準化處理，使其具有零均值和單位方差。perform_feature_engineering(df)：進行特徵工程操作，新增相關特徵以提升模型的預測能力。
特徵工程
特徵工程是提升模型表現的關鍵步驟。我們定義了一個 perform_feature_engineering 函式，用於新增相關特徵。
特徵工程函式實作
def perform_feature_engineering(df: DataFrame) -> DataFrame:
"""
進行特徵工程，新增相關特徵以提升模型的預測能力。
Parameters:
df (DataFrame): 輸入的 DataFrame。
Returns:
DataFrame: 新增特徵後的 DataFrame。
"""
df = df.withColumn('price_range', df['High'] - df['Low'])
df = df.withColumn('price_change', df['Close'] - df['Open'])
# 可根據實際需求新增其他特徵
return df
特徵工程函式解析：
df.withColumn('price_range', df['High'] - df['Low'])：新增一個欄位price_range，表示最高價與最低價之間的差異。df.withColumn('price_change', df['Close'] - df['Open'])：新增一個欄位price_change，表示收盤價與開盤價之間的差異。
