How does Spark partition(ing) work on files in HDFS?

Question

I'm working with Apache Spark on a Cluster using HDFS. As far as I understand, HDFS is distributing files on data-nodes. So if a put a "file.txt" on the filesystem, it will be split into partitions. Now I'm calling

rdd = SparkContext().textFile("hdfs://.../file.txt") 

from Apache Spark. Has rdd now automatically the same partitions as "file.txt" on the filesystem? What happens when I call

rdd.repartition(x)

where x > then the partitions used by hdfs? Will Spark physically rearrange the data on hdfs to work locally?

Example: I put a 30GB Textfile on the HDFS-System, which is distributing it on 10 nodes. Will Spark a) use the same 10 partitons? and b) shuffle 30GB across the cluster when I call repartition(1000)?

Answer 1

When Spark reads a file from HDFS, it creates a single partition for a single input split. Input split is set by the Hadoop InputFormat used to read this file. For instance, if you use textFile() it would be TextInputFormat in Hadoop, which would return you a single partition for a single block of HDFS (but the split between partitions would be done on line split, not the exact block split), unless you have a compressed text file. In case of compressed file you would get a single partition for a single file (as compressed text files are not splittable).

When you call rdd.repartition(x) it would perform a shuffle of the data from N partititons you have in rdd to x partitions you want to have, partitioning would be done on round robin basis.

If you have a 30GB uncompressed text file stored on HDFS, then with the default HDFS block size setting (128MB) it would be stored in 235 blocks, which means that the RDD you read from this file would have 235 partitions. When you call repartition(1000) your RDD would be marked as to be repartitioned, but in fact it would be shuffled to 1000 partitions only when you will execute an action on top of this RDD (lazy execution concept)

Answer 2

When reading non-bucketed HDFS files (e.g. parquet) with spark-sql, the number of DataFrame partitions df.rdd.getNumPartitions depends on these factors:

• spark.default.parallelism (roughly translates to #cores available for the application)
• spark.sql.files.maxPartitionBytes (default 128MB)
• spark.sql.files.openCostInBytes (default 4MB)

A rough estimation of the number of partitions is:

• If you have enough cores to read all your data in parallel, (i.e. at least one core for every 128MB of your data)

  AveragePartitionSize ≈ min(4MB, TotalDataSize/#cores) NumberOfPartitions ≈ TotalDataSize/AveragePartitionSize

• If you don't have enough cores,

  AveragePartitionSize ≈ 128MB NumberOfPartitions ≈ TotalDataSize/AveragePartitionSize

The exact calculations are slightly complicated and can be found on the code base for FileSourceScanExec, refer here.

Answer 3

Here is the snapshot of "How blocks in HDFS are loaded into Spark workers as partitions"

In this images 4 HDFS blocks are loaded as Spark partitions inside 3 workers memory

---

Example: I put a 30GB Textfile on the HDFS-System, which is distributing it on 10 nodes.

Will Spark

a) use the same 10 partitions?

Spark load same 10 HDFS bocks to workers memory as partitions. I assume block size of 30 GB file should be 3 GB to get 10 partitions/blocks (with default conf)

b) shuffle 30GB across the cluster when I call repartition(1000)?

Yes, Spark shuffle the data among the worker nodes in order to create 1000 partitions in workers memory.

Note:

HDFS Block -> Spark partition   : One block can represent as One partition (by default)
Spark partition -> Workers      : Many/One partitions can present in One workers 

Answer 4

Addition to @0x0FFF If it take from HDFS as input file it will calculate like for this rdd = SparkContext().textFile("hdfs://.../file.txt") and when you do rdd.getNumPatitions it will result Max(2, Number of HDFS block). I did lot of experiments and found this as result. Again explicitly you can do rdd = SparkContext().textFile("hdfs://.../file.txt", 400) to get 400 as partitions or even can do re-partitions by rdd.repartition or decrease to 10 by rdd.coalesce(10)