10

I am trying to find a fit hardware size for my Spark job. My understanding was that scaling up the number of machines may help speeding up my job, considering the fact that my job does not have a complex action operation and therefore probably small amount of calculation in the driver program. However, what I observe is that the job execution speed lowers down when adding resources to Spark. I can reproduce this effect using the following simple job:

  • Loading a text file (~100Gb) from HDFS
  • Running a simple 'filter' transformation on the RDD, that looks like below:

    JavaRDD<String> filteredRDD = rdd.filter(new Function<String, Boolean>() {
        public Boolean call(String s) {
            String filter = "FILTER_STRING";
            return s.indexOf(filter) > 0 ? true : false; 
       }
    
  • Running count() action on the result

The scaling problem shows itself when I scale up the number of machines in the cluster from 4 to 8. Here are some details about the environment:

  • Each executor is configured to use 6 GB of memory. Also the HDFS is co-hosted on the same machines.
  • Each machine has 24 GB of RAM in total and 12 cores (configured to use 8 for Spark executors).
  • Spark is hosted in a YARN cluster.

Any ideas why I am not getting the degree of scalabilty I expect from Spark?

17
  • how much partitions are there in your rdd? How many cores each machine has? memory? Please provide some details. What do you see in spark UI Mar 13, 2016 at 21:32
  • @IgorBerman I edited the question to give more on the test env. As for the number of partitions, I am letting Spark decide (which creates partitions for each HDFS block, in my case 128MB). It is extremely hard for me to access the UI, but everything is OK in the logs. All executors are launched and no errors that I can see. Should I look for something specific in the logs?
    – asaad
    Mar 13, 2016 at 22:27
  • What are the times and what kind of disk do you use?
    – zero323
    Mar 13, 2016 at 23:34
  • @zero323 I use SSD as disk. No RAID. I get 6-7 seconds responses when using 4 machines, 12 when scale to 8 and ~30 seconds when go up to 16.
    – asaad
    Mar 13, 2016 at 23:38
  • 1
    It is just a wild guess but with fast disks and trivial tasks you may be wasting all gains from increased parallelism on additional scheduling and communication. Also there could some data locality issues. One thing you can try is to artificially increase tasks cost (even with simply thread sleep ...) and see how your clusters behave then.
    – zero323
    Mar 14, 2016 at 7:47

1 Answer 1

6

Thanks to lot of comments, I think I found what was wrong with my cluster. The idea of HDFS 'replication factor' being at least a part of the problem was a very good clue.

In order to test, I changed replication factor of HDFS to the number of cluster nodes and re-ran the tests, and I got scalable results. But I was not convinced about the reason behind this behavior, because Spark claims to consider data locality in assigning partitions to executors and even with the default replication level (3), Spark should have enough room to assign partitions evenly. With some more investigation I figured out that this may not be the case if YARN (or any other cluster manager) decide to share a physical machine with more than one executor and not to use all the machines. In that case there may be HDFS blocks that are not local to any executor, which will result in data transfer across network and the scaling problem that I observed.

5
  • When you add new nodes to Hadoop you need to rebalance the HDFS so that data will spread evenly between the nodes. Mar 24, 2016 at 9:40
  • @DanielHaviv That's true. But that was not my case because I started my HDFS fresh and copied the test file in order to ensure the files are in balance.
    – asaad
    Mar 24, 2016 at 14:42
  • So is there any work around without changing replication to cover the whole cluster? Like knowing why YARN may fail assigning the wrong container to the wrong machine?
    – Maziyar
    Jun 20, 2017 at 23:28
  • 1
    @Maziyar. I couldn't find any other workarounds since I posted the question. Note that while HDFS/Spark could do a better job allocating the partitions, they are not doing it wrong! Problem is that data locality breaks when data starts shuffling around the cluster. So you (as the programmer) should have this design in mind and tune your code to minimize the shuffling of the data.
    – asaad
    Jun 26, 2017 at 13:48
  • I appreciate your answer @asaad. The problem is, most of ML and MlLib functions are forced to do the shuffling before you can go to the next step. But I will look into caching and broadcasting to see if I can make it work. (Actually it is very fast alone, only when I mix it with Stanford CoreNLP it becomes a bit slower). Thanks for your reply again.
    – Maziyar
    Jun 27, 2017 at 9:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.