深度學習模型的訓練仰賴大量的資料，而資料準備階段的效率往往是決定專案成敗的關鍵。PySpark 提供了分散式運算能力，能有效處理大規模資料集。本文以 Tesla 股票價格預測為例，示範如何使用 PySpark 從 S3 下載資料、進行初步的資料探索，並利用 PySpark 的平行處理特性最佳化資料處理流程。首先，程式碼會從指定的 S3 儲存桶下載 CSV 格式的 Tesla 股票資料到本地檔案系統。接著，利用 SparkSession 載入資料，並執行一系列的資料處理步驟，包含計算描述性統計量、檢查缺失值和繪製收盤價走勢圖等，以便快速瞭解資料集的特性。
# 無程式碼
使用 PySpark 進行深度學習的資料準備
從 S3 複製檔案到本地檔案系統
在資料處理流程中，第一步是從 Amazon S3 將檔案複製到本地檔案系統。以下是一個名為 copy_file_from_s3 的函式，負責執行此任務：
def copy_file_from_s3(bucket_name: str, file_key: str, local_file_path: str):
"""
從 S3 複製檔案到本地檔案路徑。
:param bucket_name: S3 儲存桶名稱
:param file_key: S3 檔案鍵值
:param local_file_path: 本地檔案路徑
"""
try:
s3 = boto3.client('s3')
s3.download_file(bucket_name, file_key, local_file_path)
print(f"成功將檔案從 S3://{bucket_name}/{file_key} 複製到本地檔案路徑：{local_file_path}")
except Exception as e:
print(f"從 S3 複製檔案時發生錯誤：{str(e)}")
內容解密：
- 函式定義：
copy_file_from_s3函式接受三個引數：bucket_name（S3 儲存桶名稱）、file_key（S3 檔案鍵值）和local_file_path（本地檔案路徑）。 - 錯誤處理：使用
try-except區塊處理可能發生的錯誤，確保程式在遇到問題時能夠提供有用的錯誤訊息。 - S3 使用者端建立：使用
boto3.client('s3')建立 S3 使用者端物件，用於與 Amazon S3 服務互動。 - 檔案下載：呼叫
s3.download_file()方法，將檔案從指定的 S3 儲存桶和鍵值下載到本地檔案路徑。 - 成功訊息：下載成功後，列印確認訊息，顯示來源 S3 儲存桶和目標本地檔案路徑。
- 錯誤訊息：若下載過程中發生錯誤，捕捉例外並列印錯誤訊息，以便診斷和故障排除。
主函式：初始化 SparkSession 和資料處理
主函式 main 是程式的進入點，負責初始化 SparkSession、複製檔案、載入資料並執行資料處理任務。
def main(s3_bucket_name: str, s3_file_key: str, local_file_path: str):
"""
股票價格預測的主函式。
:param s3_bucket_name: S3 儲存桶名稱
:param s3_file_key: S3 檔案鍵值
:param local_file_path: 本地檔案路徑
"""
spark = SparkSession.builder.appName("StockPricePrediction").getOrCreate()
copy_file_from_s3(s3_bucket_name, s3_file_key, local_file_path)
data_processor = DataProcessor(spark)
df = data_processor.load_data(local_file_path)
if df is not None:
data_processor.print_first_n_rows(df)
data_processor.calculate_descriptive_statistics(df)
data_processor.check_for_null_values(df)
data_processor.visualize_data(df)
else:
print("載入資料時發生錯誤，程式終止。")
內容解密：
- SparkSession 初始化：使用
SparkSession.builder方法建立 SparkSession 物件，應用程式名稱為 “StockPricePrediction”。 - 檔案複製：呼叫
copy_file_from_s3函式，將檔案從 S3 複製到本地檔案路徑。 - DataProcessor 初始化：使用建立的 SparkSession 初始化
DataProcessor物件。 - 資料載入：使用
DataProcessor的load_data方法載入本地檔案中的資料。 - 資料處理：若資料載入成功，執行多項資料處理任務，包括列印前 N 列、計算描述性統計、檢查空值和視覺化資料。
- 錯誤處理：若資料載入失敗，列印錯誤訊息並終止程式。
程式執行與引數設定
最後，程式使用一個條件區塊檢查是否直接執行該指令碼，並設定 S3 儲存桶名稱、檔案鍵值和本地檔案路徑等引數，然後呼叫 main 函式。
if __name__ == "__main__":
s3_bucket_name = 'instance1bucket'
s3_file_key = 'TSLA_stock.csv'
local_file_path = '/home/ubuntu/airflow/dags/TSLA_stock.csv'
main(s3_bucket_name, s3_file_key, local_file_path)
內容解密：
- 引數設定：定義三個變數：
s3_bucket_name、s3_file_key和local_file_path，分別代表 S3 儲存桶名稱、檔案鍵值和本地檔案路徑。 - main 函式呼叫：將上述引數傳遞給
main函式，執行整個資料處理流程。
指令碼執行
儲存指令碼為 tesla_stock_exploration.py，啟動 EC2 例項上的虛擬環境，並執行 Python 指令碼。
cd /home/ubuntu
source myenv/bin/activate
python3 tesla_stock_exploration.py
內容解密：
- 目錄切換：切換到
/home/ubuntu目錄。 - 虛擬環境啟動：啟動名為
myenv的虛擬環境。 - Python 指令碼執行：使用 Python 3 執行
tesla_stock_exploration.py指令碼。
檢視程式碼輸出結果
程式碼輸出分析
程式碼執行後產生多個輸出結果，首先是 copy_file_from_s3 函式輸出的訊息，表示 CSV 檔案已成功從 S3 儲存桶下載到本地檔案路徑：
File downloaded from S3 bucket instance1bucket to local file path: /home/ubuntu/airflow/dags/TSLA_stock.csv
這個輸出確認了檔案下載的成功與否，以及本地儲存的路徑。
資料檢視
第二個輸出是 DataFrame 的前十行，由 print_first_n_rows 方法生成，顯示了 Tesla 股票的日期（Date）、開盤價（Open）、最高價（High）、最低價（Low）、收盤價（Close）和成交量（Volume）等欄位：
Date Open High Low Close Volume
2/23/24 195.31 197.57 191.50 191.97 78,670,300
2/22/24 194.00 198.32 191.36 197.41 92,739,500
2/21/24 193.36 199.44 191.95 194.77 103,844,000
2/20/24 196.13 198.60 189.13 193.76 104,545,800
2/16/24 202.06 203.17 197.40 199.95 111,173,600
2/15/24 189.16 200.88 188.86 200.45 120,831,800
2/14/24 185.30 188.89 183.35 188.71 81,203,000
2/13/24 183.99 187.26 182.11 184.02 86,759,500
2/12/24 192.11 194.73 187.28 188.13 95,498,600
2/09/24 190.18 194.12 189.48 193.57 84,476,300
這個輸出提供了資料集的初步概覽，有助於快速瞭解資料的結構和內容是否正確。
程式碼解密：
- 輸出前十行資料的作用：透過列印 DataFrame 的前十行，可以直觀檢查資料的完整性，確認資料是否正確匯入，並初步瞭解資料的數值範圍。
- 欄位意義：日期（Date）、開盤價（Open）、最高價（High）、最低價（Low）、收盤價（Close）和成交量（Volume）是股票資料分析中的關鍵欄位，這些資料對於後續的技術分析和模型訓練至關重要。
- 檢查資料正確性：輸出結果可以幫助檢查資料是否正確載入，例如檢查日期是否連續、價格資料是否有異常波動等。
描述性統計分析
第三個輸出是資料集中各數值欄位的描述性統計結果，由 calculate_descriptive_statistics 方法生成：
Summary Open High Low Close Volume
count 1,258 1,258 1,258 1,258 1,258
mean 176.37 180.31 172.10 176.31 133,933,068
stddev 105.44 107.68 102.90 105.28 85,052,921
min 12.07 12.45 11.80 11.93 29,401,800
25% 57.20 59.10 55.59 57.63 81,203,000
50% 202.59 208.00 198.50 203.33 109,536,700
75% 251.45 256.59 246.35 251.92 157,577,100
max 411.47 414.50 405.67 409.97 914,082,000
這些統計資料提供了對資料集的全面瞭解，包括集中趨勢、離散程度和分佈情況。
程式碼解密：
統計指標解釋：
- count：表示每個數值欄位的觀測值數量，全部為1,258，表明資料集在這些欄位上沒有缺失值。
- mean：各欄位的平均值，例如開盤價（Open）的平均值為176.37，收盤價（Close）的平均值為176.31。
- stddev：各欄位的標準差，表示資料的離散程度，例如開盤價的標準差為105.44，表明其波動較大。
- min 和 max：各欄位的最小值和最大值，用於檢查是否有異常資料。
- 25%、50%、75%：分別表示第一四分位數（Q1）、中位數（Median）和第三四分位數（Q3），用於瞭解資料的分佈情況。
分析資料特點：透過這些統計量，可以初步判斷資料的分佈特徵和變異情況。例如，成交量（Volume）的標準差較大，表明其波動範圍較廣。
缺失值檢查
程式碼還輸出了缺失值檢查的結果，由 check_for_null_values 方法生成：
Date Open High Low Close Volume
0 0 0 0 0 0
結果顯示，所有欄位均無缺失值，這對於後續的資料分析和建模至關重要。
程式碼解密：
- 缺失值處理的重要性：在機器學習中，缺失值會對模型效能和準確性產生重大影響。常見的處理方法包括均值/中位數填補、預測填補和刪除策略。
- 當前資料集的狀況：由於 Tesla 股票價格資料集中沒有缺失值，因此不需要進行缺失值處理，這簡化了資料預處理步驟，提高了後續分析的可靠性。
資料視覺化
最後，visualize_data 方法生成了 Tesla 股票收盤價隨時間變化的圖表（圖3-1）。
@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle
title PySpark 深度學習資料準備與平行處理最佳化
package "機器學習流程" {
package "資料處理" {
component [資料收集] as collect
component [資料清洗] as clean
component [特徵工程] as feature
}
package "模型訓練" {
component [模型選擇] as select
component [超參數調優] as tune
component [交叉驗證] as cv
}
package "評估部署" {
component [模型評估] as eval
component [模型部署] as deploy
component [監控維護] as monitor
}
}
collect --> clean : 原始資料
clean --> feature : 乾淨資料
feature --> select : 特徵向量
select --> tune : 基礎模型
tune --> cv : 最佳參數
cv --> eval : 訓練模型
eval --> deploy : 驗證模型
deploy --> monitor : 生產模型
note right of feature
特徵工程包含：
- 特徵選擇
- 特徵轉換
- 降維處理
end note
note right of eval
評估指標：
- 準確率/召回率
- F1 Score
- AUC-ROC
end note
@enduml此圖示顯示了 Tesla 股票收盤價的時間序列變化，有助於直觀瞭解股價的整體趨勢和波動情況。
圖示解說：
- 圖表功能：透過視覺化收盤價，可以直觀觀察到股價的變化趨勢、波動性和可能的異常點。
- 分析價值：這類別圖表對於技術分析、趨勢預測和投資決策具有重要參考價值。
PySpark 平行處理最佳化技術解析
在深度學習專案中，高效的資料處理是成功的關鍵。PySpark 提供了強大的平行處理能力，能夠顯著提升大規模資料集的處理效率。本文將探討 PySpark 的平行處理機制，並透過 Tesla 股價資料集示範如何最佳化資料分佈。
為何需要平行處理？
在處理大規模資料集時，傳統的單執行緒處理方式往往效率低下。PySpark 的平行處理機制透過將資料分割成多個分割槽（partitions），並在多個核心或節點上平行處理，顯著提高了運算效率。這種分散式處理能力使得 PySpark 成為處理海量資料的理想選擇。
資料分割槽技術詳解
資料分割槽是 PySpark 平行處理的核心。適當的資料分割槽策略可以最佳化資料處理效能。以下程式碼展示瞭如何檢查和調整 DataFrame 的分割槽數量：
class DataProcessor:
def __init__(self, spark_session):
self.spark = spark_session
def print_partition_info(self, df):
"""顯示 DataFrame 的分割槽數量"""
num_partitions = df.rdd.getNumPartitions()
print(f"分割槽數量: {num_partitions}")
內容解密：
df.rdd.getNumPartitions()用於取得 DataFrame 目前的分割槽數量。- 透過
print_partition_info方法，我們可以在重新分割槽前後監控分割槽數量的變化。 - 這有助於瞭解資料是如何在不同分割槽之間分佈的。
重新分割槽操作實務
重新分割槽是最佳化資料分佈的關鍵步驟。以下程式碼示範瞭如何將 DataFrame 重新分割槽到指定的分割槽數量：
def main(s3_bucket_name: str, s3_file_key: str, local_file_path: str):
# 初始化 SparkSession
spark = SparkSession.builder.appName("StockPriceRepartitioning").getOrCreate()
# 初始化 DataProcessor
data_processor = DataProcessor(spark)
# 載入資料
df = data_processor.load_data(local_file_path)
if df is not None:
# 顯示初始分割槽資訊
data_processor.print_partition_info(df)
# 重新分割槽到 10 個分割槽
repartitioned_df = df.repartition(10)
print("重新分割槽後:")
data_processor.print_partition_info(repartitioned_df)
# 顯示重新分割槽後的資料前幾行
data_processor.print_first_n_rows(repartitioned_df)
else:
print("資料載入失敗。")
內容解密：
df.repartition(10)將 DataFrame 重新分割槽到 10 個分割槽。- 重新分割槽前後都呼叫了
print_partition_info方法，以觀察分割槽數量的變化。 - 這種重新分割槽操作可以根據叢集的組態和資料量進行調整，以達到最佳的效能。
完整程式碼實作
以下是完整的程式碼，展示瞭如何結合 S3 檔案下載和 PySpark 資料處理：
import boto3
from pyspark.sql import SparkSession
class DataProcessor:
# ... (類別實作細節)
def copy_file_from_s3(bucket_name: str, file_key: str, local_file_path: str):
"""從 S3 下載檔案到本地"""
try:
s3 = boto3.client('s3')
s3.download_file(bucket_name, file_key, local_file_path)
print(f"檔案已下載到: {local_file_path}")
except Exception as e:
print(f"下載檔案失敗: {str(e)}")
def main(s3_bucket_name: str, s3_file_key: str, local_file_path: str):
# ... (main 函式實作細節)
if __name__ == "__main__":
s3_bucket_name = 'instance1bucket'
s3_file_key = 'TSLA_stock.csv'
local_file_path = '/home/ubuntu/airflow/dags/TSLA_stock.csv'
main(s3_bucket_name, s3_file_key, local_file_path)
