了解 Spark 创建的分区数量

Question

读取 .csv 时 pyspark-sql 将创建多少个分区？

我对此的理解是 number of partitions = math.ceil(file_size/spark.conf.get('spark.sql.files.maxPartitionBytes'))

在我的机器上：

spark.conf.get('spark.sql.files.maxPartitionBytes')
output: 
'134217728b' #128MBs

但是，我没有观察到这种行为。我创建了一个在磁盘上占用 96 MB 的文件。我在本地模式下运行 Spark。我有一台 8 核笔记本电脑。我认为它应该读入 1 个分区。但是，该文件在 8 个分区中被读取。以下是我使用过的代码库：

import pandas as pd
import numpy as np
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
#creating a small DataFrame. This will occupy 96 MBs on disk
pd.DataFrame({'id':np.arange(10000000),'b':np.random.choice(['a','b','c','d'],size=(10000000,),p=[0.25,0.25,0.25,0.25])}).to_csv('df_s.csv',index=None)
sd=spark.read.csv('df_s.csv',schema="id int, b string")
sd.rdd.getNumPartitions()
output: 8

您能帮我理解为什么无论文件大小如何我都会看到 8 个分区吗？

Answer 1

实际的公式实际上比这更复杂一些。检查下面的计算。您可以在这里找到源代码。

这是您的配置和文件

火花配置	价值	默认
Spark.sql.files.maxPartitionBytes	128M	128M
Spark.sql.files.openCostInBytes	4M	4M
Spark.executor.instances	1	当地的
Spark.executor.cores	8	你的核心
Spark.默认并行度	8	=spark.executor.instances * Spark.executor.cores
数据文件大小	64M
数据文件计数	1

这是实际的公式

	公式	字节
默认最大分割字节	= Spark.sql.files.maxPartitionBytes	134,217,728
打开成本（以字节为单位）	= Spark.sql.files.openCostInBytes	4,194,304
默认并行度	= Spark.默认.并行度	8
总字节数	= DataBytes + (# 文件 * OpenCostInBytes)	71,303,168
每核字节数	= 总字节数 / 默认并行度	8,912,896
最大分割字节数	= MIN(DefaultMaxSplitBytes, MAX(OpenCostInBytes, BytesPerCore))	8,912,896
预计分区数	= 总字节数 / 最大分割字节数	8