理解Spark写入API的数据处理能力-不念博客

下方图片解释了Apache Spark DataFrame写入API的流程。

它始于对写入数据的API调用，支持的格式包括 CSV、JSON 或 Parquet。

流程根据选择的保存模式（追加、覆盖、忽略或报错）而分岔。

每种模式执行必要的检查和操作，例如分区和数据写入处理。

流程以数据的最终写入或错误结束，取决于这些检查和操作的结果。

Apache Spark 是一个开源的分布式计算系统，提供了强大的平台用于处理大规模数据。

写入 API 是 Spark 数据处理能力的基本组成部分，允许用户将数据从他们的 Spark 应用程序写入或输出到不同的数据源。

理解 Spark 写入 API

数据源

Spark 支持将数据写入各种数据源，包括但不限于：

•分布式文件系统，如 HDFS•云存储，如 AWS S3、Azure Blob Storage•传统数据库（包括 SQL 和 NoSQL）•大数据文件格式（Parquet、Avro、ORC）

DataFrameWriter

写入 API 的核心类是 DataFrameWriter，它提供配置和执行写入操作的功能。

通过在 DataFrame 或 Dataset 上调用 .write 方法获得 DataFrameWriter。

写入模式

指定 Spark 在写入数据时应如何处理现有数据的模式。

常见的模式包括：

append：将新数据添加到现有数据中。
overwrite：用新数据覆盖现有数据。
ignore：如果数据已存在，则忽略写入操作。
errorIfExists（默认）：如果数据已存在，则抛出错误。

格式规范

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

分区

为了实现有效的数据存储，可以使用 .partitionBy("column") 方法根据一个或多个列对输出数据进行分区。

配置选项

可以使用 .option("key", "value") 方法设置特定于数据源的各种选项，如压缩、CSV 文件的自定义分隔符等。

保存数据

最后，使用 .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()