理解 Spark 寫入 API 的數據處理能力-魔扣目錄

理解 Spark 寫入 API 的數據處理能力

這張圖解釋了 Apache Spark DataFrame 寫入 API 的流程。它始于對寫入數據的 API 調用，支持的格式包括 CSV、JSON 或 Parquet。流程根據選擇的保存模式（追加、覆蓋、忽略或報錯）而分岔。每種模式執行必要的檢查和操作，例如分區和數據寫入處理。流程以數據的最終寫入或錯誤結束，取決于這些檢查和操作的結果。

Apache Spark 是一個開源的分布式計算系統，提供了強大的平臺用于處理大規模數據。寫入 API 是 Spark 數據處理能力的基本組成部分，允許用戶將數據從他們的 Spark 應用程序寫入或輸出到不同的數據源。

一、理解 Spark 寫入 API

1.數據源

Spark 支持將數據寫入各種數據源，包括但不限于：

分布式文件系統，如 HDFS
云存儲，如 AWS S3、Azure Blob Storage
傳統數據庫（包括 SQL 和 NoSQL）
大數據文件格式（Parquet、Avro、ORC）

2.DataFrameWriter

寫入 API 的核心類是 DataFrameWriter。它提供配置和執行寫入操作的功能。通過在 DataFrame 或 Dataset 上調用 .write 方法獲得 DataFrameWriter。

3.寫入模式

指定 Spark 在寫入數據時應如何處理現有數據的模式。常見的模式包括：

Append：將新數據添加到現有數據中。
overwrite：用新數據覆蓋現有數據。
ignore：如果數據已存在，則忽略寫入操作。
errorIfExists（默認）：如果數據已存在，則拋出錯誤。

4.格式規范

可以使用 .format("formatType") 方法指定輸出數據的格式，如 JSON、CSV、Parquet 等。

5.分區

為了實現有效的數據存儲，可以使用 .partitionBy("column") 方法根據一個或多個列對輸出數據進行分區。

6.配置選項

可以使用 .option("key", "value") 方法設置特定于數據源的各種選項，如壓縮、CSV 文件的自定義分隔符等。

7.保存數據

最后，使用 .save("path") 方法將 DataFrame 寫入指定的路徑。其他方法如 .saveAsTable("tableName") 也可用于不同的寫入場景。

from pyspark.sql import SparkSession
from pyspark.sql import Row
import os

# 初始化 SparkSession
spark = SparkSession.builder  
    .appName("DataFrameWriterSaveModesExample")  
    .getOrCreate()

# 示例數據
data = [
    Row(name="Alice", age=25, country="USA"),
    Row(name="Bob", age=30, country="UK")
]

# 附加數據用于追加模式
additional_data = [
    Row(name="Carlos", age=35, country="SpAIn"),
    Row(name="Daisy", age=40, country="Australia")
]

# 創建 DataFrames
df = spark.createDataFrame(data)
additional_df = spark.createDataFrame(additional_data)

# 定義輸出路徑
output_path = "output/csv_save_modes"

# 函數：列出目錄中的文件
def list_files_in_directory(path):
    files = os.listdir(path)
    return files

# 顯示初始 DataFrame
print("初始 DataFrame:")
df.show()

# 使用覆蓋模式寫入 CSV 格式
df.write.csv(output_path, mode="overwrite", header=True)
print("覆蓋模式后的文件:", list_files_in_directory(output_path))

# 顯示附加 DataFrame
print("附加 DataFrame:")
additional_df.show()

# 使用追加模式寫入 CSV 格式
additional_df.write.csv(output_path, mode="append", header=True)
print("追加模式后的文件:", list_files_in_directory(output_path))

# 使用忽略模式寫入 CSV 格式
additional_df.write.csv(output_path, mode="ignore", header=True)
print("忽略模式后的文件:", list_files_in_directory(output_path))

# 使用 errorIfExists 模式寫入 CSV 格式
try:
    additional_df.write.csv(output_path, mode="errorIfExists", header=True)
except Exception as e:
    print("errorIfExists 模式中發生錯誤:", e)

# 停止 SparkSession
spark.stop()

二、Spark 架構概述

理解 Spark 寫入 API 的數據處理能力

在 Apache Spark 中寫入 DataFrame 遵循一種順序流程。Spark 基于用戶 DataFrame 操作創建邏輯計劃，優化為物理計劃，并分成階段。系統按分區處理數據，對其進行日志記錄以確保可靠性，并帶有定義的分區和寫入模式寫入到本地存儲。Spark 的架構確保在計算集群中高效管理和擴展數據寫入任務。

從 Spark 內部架構的角度來看，Apache Spark 寫入 API 涉及了解 Spark 如何在幕后管理數據處理、分發和寫入操作。讓我們來詳細了解：

三、Spark 架構概述

驅動程序和執行器： Spark 采用主從架構。驅動節點運行應用程序的 main() 函數并維護有關 Spark 應用程序的信息。執行器節點執行數據處理和寫入操作。
DAG 調度器：當觸發寫入操作時，Spark 的 DAG（有向無環圖）調度器將高級轉換轉換為一系列可以在集群中并行執行的階段。
任務調度器：任務調度器在每個階段內啟動任務。這些任務分布在執行器之間。
執行計劃和物理計劃： Spark 使用 Catalyst 優化器創建高效的執行計劃。這包括將邏輯計劃（要做什么）轉換為物理計劃（如何做），考慮到分區、數據本地性和其他因素。