理解Spark写入API的数据处理能力

理解Spark写入API的数据处理能力

一篇关于理解Spark写入API的数据处理能力IT技巧方面的经验，下面小编为您详细解答

下方图片解释了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()