为什么CSV会根据文件大小读取Apache Spark分区以及如何更改分区？

Question

下面是我的pyspark代码：

csv_file = "/FileStore/tables/mnt/training/departuredelays02.csv"
schema   = "`date` STRING, `delay` INT, `distance` INT, `origin` STRING, `destination` STRING"
df = (spark
     .read
     .format("csv")                    
     .option("header","true")
     .schema(schema)
     .load(csv_file)                  
        )
partitions = df.rdd.getNumPartitions()
print(partitions)

现在，如果我尝试加载以下文件，该文件大得多，有1391578行：

csv_file = "/FileStore/tables/mnt/training/departuredelays.csv"

我得到一个8的分区。

我的问题是如何强制第一个CSV以与较大文件相同的方式进行分区。我知道可以使用重新分区，但我很想知道这是否可以在没有任何洗牌的情况下完成？而且即使我们重新划分它，它似乎创建了一个有3个任务而不是8个任务的作业。

df = df.repartition(8)
print(df.count())

(3) Spark Jobs
Job 93 View(Stages: 1/1)
Stage 123: 3/3
Job 94 View(Stages: 1/1, 1 skipped)
Stage 124: 0/3 skipped
Stage 125: 8/8
Job 95 View(Stages: 1/1, 2 skipped)
Stage 126: 0/3 skipped
Stage 127: 0/8 skipped
Stage 128: 1/1

Answer 1

您可以尝试使用coalesce，这样可以进行合理的洗牌，而不是重新分区。

df = spark
     .read
     .format("csv")                    
     .option("header","true")
     .schema(schema)
     .load(csv_file)                  
     .coalesce(8)

有关Spark-repartition（）vs coalesce（）的更多信息，请参见此