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

apache-spark hdfs
Degget picture Degget · Mar 12, 2015 · Viewed 42.7k times · Source

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

0x0FFF picture 0x0FFF · Mar 12, 2015

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)

Related questions

Spark - load CSV file as DataFrame?

I would like to read a CSV in spark and convert it as DataFrame and store it in HDFS with df.registerTempTable("table_name") I have tried: scala> val df = sqlContext.load("hdfs:///csv/file/dir/file.csv") Error …

Read More
SparkSQL - Read parquet file directly

I am migrating from Impala to SparkSQL, using the following code to read a table: my_data = sqlContext.read.parquet('hdfs://my_hdfs_path/my_db.db/my_table') How do I invoke SparkSQL above, so it can return something …

Read More
Write to multiple outputs by key Spark - one Spark job

How can you write to multiple outputs dependent on the key using Spark in a single Job. Related: Write to multiple outputs by key Scalding Hadoop, one MapReduce Job E.g. sc.makeRDD(Seq((1, "a"), (1, "b"), (2, "c"))) .writeAsMultiple(prefix, compressionCodecOption) …

Read More