PySpark大数据存储问题咨询：求多机器分存方案及代码示例

方案可行性说明 #

• 方案1完全可行：PySpark基于分布式架构设计，天然支持加载分散在集群不同节点的分片数据，只要集群内节点能访问到这些数据的存储路径即可。你可以将大数据集拆分为多个单节点可容纳的小子集，分别存储在集群不同节点的可访问路径（如本地磁盘、共享存储），PySpark会自动并行加载这些分片进行后续处理。
• 方案2答案是肯定的：HDFS本身就是分布式存储系统，数据上传后会自动被切分为固定大小的数据块（默认128MB），分散存储在集群的多个DataNode节点上，无需额外配置就能利用多台机器的存储资源。

方案1的示例代码 #

场景1：结构化分片文件（CSV/Parquet）批量加载 #

假设你已将大文件拆分为多个CSV分片，存储在集群共享路径（如NFS挂载的/data/shards/）或各节点本地路径：

from pyspark.sql import SparkSession

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

# 用通配符匹配所有分片文件，Spark会自动并行加载
# 共享存储路径示例
df = spark.read.csv("/data/shards/shard_*.csv", header=True, inferSchema=True)

# 节点本地存储路径示例（需确保executor能访问到对应节点的本地文件）
# df = spark.read.csv("file:///local/data/shards/shard_*.csv", header=True, inferSchema=True)

# 后续处理：示例为过滤+聚合
processed_df = df.filter(df["amount"] > 100).groupBy("region").sum("amount")

# 分布式写入结果
processed_df.write.parquet("/output/aggregated_result.parquet", mode="overwrite")

spark.stop()

场景2：自定义格式分片手动指定路径加载 #

如果分片是自定义格式（如二进制文件、特殊文本格式），可手动列出所有分片路径让Spark并行处理：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("CustomShardLoading") \
    .getOrCreate()

# 列出所有分片的路径，可混合共享存储和节点本地路径
shard_paths = [
    "/data/shards/shard_01.dat",
    "/data/shards/shard_02.dat",
    "file:///node-02/local/data/shard_03.dat",
    "file:///node-03/local/data/shard_04.dat"
]

# 并行读取分片（此处以文本格式为例，自定义格式可替换为binaryFiles等）
shard_rdd = spark.sparkContext.textFile(','.join(shard_paths))

# 转换为DataFrame并处理
df = shard_rdd.map(lambda line: line.split('|')).toDF(["user_id", "action", "timestamp"])
df.show()

spark.stop()

关键注意事项 #

• 确保集群所有节点（Driver和Executor）都能访问分片路径：本地存储需保证对应节点有分片文件，或通过共享存储挂载统一路径；
• 分片大小建议接近Spark默认分区大小（128MB左右），以最大化并行处理效率；
• 若分片为压缩格式，PySpark支持gzip、snappy等常见压缩算法，无需额外配置即可直接加载。

内容的提问来源于stack exchange，提问作者Geoffrey