60T sadly isn't that much. We tried (and failed) to use spark on a several PB dataset and it failed miserably.
Apache Spark Scale: A 60 TB+ production use case
31–40 of 42 posts
Re: Apache Spark Scale: A 60 TB+ production use case
#32I tried (and failed after a month of work) earlier this year trying to do ~20TB shuffles with Spark. I felt both relief and frustration reading this post. Relief: I'm not an idiot, and the problems in the shuffle were likely in Spark and not just me being a beginner user. Frustration: I wanted to group about 100 billion x 200 byte records into 5 billion groups. This seems like exactly the problem Spark is designed fo…
One of the hardest parts for me was getting the cluster sized appropriately so that all data stayed in memory. Overflow to disk slows things down a lot. But, sizing the cluster can be tricky if you're generating a lot of data structures in the tasks.
I only use RDDs, so I put a lot of thought into the processing flow so that unnecessary data wasn't shuffled. If you're reducing PairRDDs, make sure that the data is evenly distributed. Also, I'm guessing you read the optimization docs, but a huge amount of network I/O can be reduced by choosing the right types and collections and optimizing serialization. And, of course group within partitions first, then within nodes, then across nodes. And, of course, go for fewer bigger servers with lots of network bandwidth.
There are a lot of tricks, unfortunately. And, since I don't know your experience level I won't bore you with things you probably already know.
Re: Apache Spark Scale: A 60 TB+ production use case
#33This is very intriguing, I cannot help but wonder how this operation would have performed on something like Google Big Query. I know that it is highly unlikely that Facebook would ever load their data to the Google Cloud Platform, but it would be an interesting comparison.
While both Spark and BigQuery do the shuffle step in-memory, there are some differences[1]: - BigQuery's execution is pipelined (don't wait for step 1 to finish to start step 2) - BigQuery's in-memory shuffler is not co-tenant to compute And, of course, it's one thing to have software and hardware. BigQuery provides a fully-managed, fully-encrypted, HA, redundant, and constantly seamlessly maintaned and upgraded serv…
Would pipelining help much when the processing job is CPU bound (all cores maxed out)?
Sorry - what does co-tenant to compute mean?
Re: Apache Spark Scale: A 60 TB+ production use case
#34Earlier quoted context omitted.
While both Spark and BigQuery do the shuffle step in-memory, there are some differences[1]: - BigQuery's execution is pipelined (don't wait for step 1 to finish to start step 2) - BigQuery's in-memory shuffler is not co-tenant to compute And, of course, it's one thing to have software and hardware. BigQuery provides a fully-managed, fully-encrypted, HA, redundant, and constantly seamlessly maintaned and upgraded serv…
Would pipelining help much when the processing job is CPU bound (all cores maxed out)? Sorry - what does co-tenant to compute mean?
Pipelining helps with hanging chads, tail latency of work steps. If you have a slow worker (due to, say, data skew), entire job slows down. All other workers are sitting idle, waiting for the one worker to finish their piece. Read [0] to see what Dataflow does, and BigQuery/Dremel do very similar stuff to deal with this issue. BigQuery also doesn't have to wait for ALL workers to finish step 1 before proceeding to step 2.
By co-tenant to compute, I mean that processing nodes themselves handle the shuffle in Spark. This can cause non-obvious bottlenecks. BigQuery handles shuffle outside of the processing nodes [1].
[0] https://cloud.google.com/blog/big-data/2016/05/no-shard-left...
[1] https://cloud.google.com/blog/big-data/2016/08/in-memory-que...
Re: Apache Spark Scale: A 60 TB+ production use case
#35I tried (and failed after a month of work) earlier this year trying to do ~20TB shuffles with Spark. I felt both relief and frustration reading this post. Relief: I'm not an idiot, and the problems in the shuffle were likely in Spark and not just me being a beginner user. Frustration: I wanted to group about 100 billion x 200 byte records into 5 billion groups. This seems like exactly the problem Spark is designed fo…
One of the hardest parts for me was getting the cluster sized appropriately so that all data stayed in memory. Overflow to disk slows things down a lot. But, sizing the cluster can be tricky if you're generating a lot of data structures in the tasks. I only use RDDs, so I put a lot of thought into the processing flow so that unnecessary data wasn't shuffled. If you're reducing PairRDDs, make sure that the data is eve…
I have no practical experience with Spark, but I was wondering where Alluxio fits into what you're describing.
Is it only applicable once the "cluster [is] sized appropriately so that all data stayed in memory" as you mention?
Re: Apache Spark Scale: A 60 TB+ production use case
#36"While running on 20 TB of input, we discovered that we were generating too many output files (each sized around 100 MB) due to the large number of tasks." I could be completely missing something here, but to decrease the number of output files you can coalesce() the RDD before you write. For example, let's say you have a 20 node cluster, each with 10 executors, and the RDD action is split into 20000 tasks. You may e…
Yeah I didn't quite get that part either.
Re: Apache Spark Scale: A 60 TB+ production use case
#3760T sadly isn't that much. We tried (and failed) to use spark on a several PB dataset and it failed miserably.
Do you have anymore details? What version?
Re: Apache Spark Scale: A 60 TB+ production use case
#38Earlier quoted context omitted.
One of the hardest parts for me was getting the cluster sized appropriately so that all data stayed in memory. Overflow to disk slows things down a lot. But, sizing the cluster can be tricky if you're generating a lot of data structures in the tasks. I only use RDDs, so I put a lot of thought into the processing flow so that unnecessary data wasn't shuffled. If you're reducing PairRDDs, make sure that the data is eve…
I have no practical experience with Spark, but I was wondering where Alluxio fits into what you're describing. Is it only applicable once the "cluster [is] sized appropriately so that all data stayed in memory" as you mention?
I think Alluxio helps with the initial data read in the first stage, but subsequent stages' datasets are stored in memory anyway.
Re: Apache Spark Scale: A 60 TB+ production use case
#39"While running on 20 TB of input, we discovered that we were generating too many output files (each sized around 100 MB) due to the large number of tasks." I could be completely missing something here, but to decrease the number of output files you can coalesce() the RDD before you write. For example, let's say you have a 20 node cluster, each with 10 executors, and the RDD action is split into 20000 tasks. You may e…
"Remove the two temporary tables and combine all three Hive stages into a single Spark job that reads 60 TB of compressed data and performs a 90 TB shuffle and sort." "As far as we know, this is the largest real-world Spark job attempted in terms of shuffle data size" I'm far, far from a world class engineer, but I regularly do 90 TiB shuffle sorts. I must seriously be missing something, here.
Have you run into any of the issues mentioned of the article? Some of them are regressions, which version of Spark were you running?
Out of the linked issues these all seem like they would be "easy" to hit given enough data:
https://issues.apache.org/jira/browse/SPARK-13279
https://issues.apache.org/jira/browse/SPARK-13850
Re: Apache Spark Scale: A 60 TB+ production use case
#40Earlier quoted context omitted.
One of the hardest parts for me was getting the cluster sized appropriately so that all data stayed in memory. Overflow to disk slows things down a lot. But, sizing the cluster can be tricky if you're generating a lot of data structures in the tasks. I only use RDDs, so I put a lot of thought into the processing flow so that unnecessary data wasn't shuffled. If you're reducing PairRDDs, make sure that the data is eve…
I have no practical experience with Spark, but I was wondering where Alluxio fits into what you're describing. Is it only applicable once the "cluster [is] sized appropriately so that all data stayed in memory" as you mention?
Alluxio can cache objects in memory that are used downstream in other steps or even other jobs.
In Spark, the biggest performance gains are realized when all data stays in memory. But, the Spark memory architecture is kind of opaque. It's pretty well documented, and there are some good blogs out there, but there are a lot of factors in play.
For example, let's say we start with 1 TB of compressed data and take a guess that it will uncompress to 10 TB. So, we do a naive calculation and guess that we'll need 20 nodes, each with 500 GB of memory.
For our example, Spark will be managed by Yarn. Yarn is going to take a portion of each node's memory. I forget how much, but let's say 10%. Then, depending on how many executors we have per node, each will also have memory overhead. Again, I don't know exactly, but let's just say another 10%. Then, some memory will be dedicated to system processes, etc. Then there's the Spark driver. So, after all of that, let's say 25-30% of memory is used outside of the executor. So, that leaves us with 10 executors per node, each with ~35 GB of memory. For maximum performance, we have to ensure that all 10 TB fits across all executors. So, we may actually need 29 nodes, or a lot of data is going to spill to disk.
Then, let's say that our processing job doesn't evenly distribute data across partitions. Some partitions get 5x as much data as other partitions. Again, data spills to disk - or the VM runs out of heap. Or, maybe our 10% compression estimate is off by 5% - more spills to disk.