[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[ https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14948943#comment-14948943 ] Glenn Strycker commented on SPARK-11004: True, fixing the 2GB will go a long way. However, this isn't a bug ticket, but a request for a new Spark feature in future versions, that MapReduce could be run on RDDs. This would add a lot of functionality for many possible use cases. This request is asking for an alternate back-end for the join operation, so if the normal rdd.join operation fails, a developer could force using this alternate MapReduce method that would run on disk rather than memory. The reason for this request is that there are actually scenarios where MapReduce really is preferable to Spark, but it is inefficient for developers to mix code and have to run programs with scripts. The request here is for additional back-end functionality for basic RDDs operations that fail at scale for Spark but work correctly in MapReduce. > MapReduce Hive-like join operations for RDDs > > > Key: SPARK-11004 > URL: https://issues.apache.org/jira/browse/SPARK-11004 > Project: Spark > Issue Type: New Feature >Reporter: Glenn Strycker > > Could a feature be added to Spark that would use disk-only MapReduce > operations for the very largest RDD joins? > MapReduce is able to handle incredibly large table joins in a stable, > predictable way with gracious failures and recovery. I have applications > that are able to join 2 tables without error in Hive, but these same tables, > when converted into RDDs, are unable to join in Spark (I am using the same > cluster, and have played around with all of the memory configurations, > persisting to disk, checkpointing, etc., and the RDDs are just too big for > Spark on my cluster) > So, Spark is usually able to handle fairly large RDD joins, but occasionally > runs into problems when the tables are just too big (e.g. the notorious 2GB > shuffle limit issue, memory problems, etc.) There are so many parameters to > adjust (number of partitions, number of cores, memory per core, etc.) that it > is difficult to guarantee stability on a shared cluster (say, running Yarn) > with other jobs. > Could a feature be added to Spark that would use disk-only MapReduce commands > to do very large joins? > That is, instead of myRDD1.join(myRDD2), we would have a special operation > myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run > MapReduce, and then convert the results of the join back into a standard RDD. > This might add stability for Spark jobs that deal with extremely large data, > and enable developers to mix-and-match some Spark and MapReduce operations in > the same program, rather than writing Hive scripts and stringing together > Spark and MapReduce programs, which has extremely large overhead to convert > RDDs to Hive tables and back again. > Despite memory-level operations being where most of Spark's speed gains lie, > sometimes using disk-only may help with stability! -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[ https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14948920#comment-14948920 ] Sean Owen commented on SPARK-11004: --- Spark has had a sort-based shuffle for a while, which is a lot of the battle. Yes there's already a known issue about 2GB block size limits. Is there a specific issue here beyond what's already in JIRA? > MapReduce Hive-like join operations for RDDs > > > Key: SPARK-11004 > URL: https://issues.apache.org/jira/browse/SPARK-11004 > Project: Spark > Issue Type: New Feature >Reporter: Glenn Strycker > > Could a feature be added to Spark that would use disk-only MapReduce > operations for the very largest RDD joins? > MapReduce is able to handle incredibly large table joins in a stable, > predictable way with gracious failures and recovery. I have applications > that are able to join 2 tables without error in Hive, but these same tables, > when converted into RDDs, are unable to join in Spark (I am using the same > cluster, and have played around with all of the memory configurations, > persisting to disk, checkpointing, etc., and the RDDs are just too big for > Spark on my cluster) > So, Spark is usually able to handle fairly large RDD joins, but occasionally > runs into problems when the tables are just too big (e.g. the notorious 2GB > shuffle limit issue, memory problems, etc.) There are so many parameters to > adjust (number of partitions, number of cores, memory per core, etc.) that it > is difficult to guarantee stability on a shared cluster (say, running Yarn) > with other jobs. > Could a feature be added to Spark that would use disk-only MapReduce commands > to do very large joins? > That is, instead of myRDD1.join(myRDD2), we would have a special operation > myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run > MapReduce, and then convert the results of the join back into a standard RDD. > This might add stability for Spark jobs that deal with extremely large data, > and enable developers to mix-and-match some Spark and MapReduce operations in > the same program, rather than writing Hive scripts and stringing together > Spark and MapReduce programs, which has extremely large overhead to convert > RDDs to Hive tables and back again. > Despite memory-level operations being where most of Spark's speed gains lie, > sometimes using disk-only may help with stability! -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[ https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14949060#comment-14949060 ] Sean Owen commented on SPARK-11004: --- Literally run a Mapper and Reducer on Spark? I think it would be more efficient to run then via MapReduce. I imagine that most people using MR still have a Hadoop cluster handy that they're running Spark on too, so this is quite viable. What's the MapReduce method you're referring to? not literally calling to MapReduce right? I'm trying to isolate the distinct change being described here. Otherwise I think this is already covered in existing improvements to Spark itself. > MapReduce Hive-like join operations for RDDs > > > Key: SPARK-11004 > URL: https://issues.apache.org/jira/browse/SPARK-11004 > Project: Spark > Issue Type: New Feature >Reporter: Glenn Strycker > > Could a feature be added to Spark that would use disk-only MapReduce > operations for the very largest RDD joins? > MapReduce is able to handle incredibly large table joins in a stable, > predictable way with gracious failures and recovery. I have applications > that are able to join 2 tables without error in Hive, but these same tables, > when converted into RDDs, are unable to join in Spark (I am using the same > cluster, and have played around with all of the memory configurations, > persisting to disk, checkpointing, etc., and the RDDs are just too big for > Spark on my cluster) > So, Spark is usually able to handle fairly large RDD joins, but occasionally > runs into problems when the tables are just too big (e.g. the notorious 2GB > shuffle limit issue, memory problems, etc.) There are so many parameters to > adjust (number of partitions, number of cores, memory per core, etc.) that it > is difficult to guarantee stability on a shared cluster (say, running Yarn) > with other jobs. > Could a feature be added to Spark that would use disk-only MapReduce commands > to do very large joins? > That is, instead of myRDD1.join(myRDD2), we would have a special operation > myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run > MapReduce, and then convert the results of the join back into a standard RDD. > This might add stability for Spark jobs that deal with extremely large data, > and enable developers to mix-and-match some Spark and MapReduce operations in > the same program, rather than writing Hive scripts and stringing together > Spark and MapReduce programs, which has extremely large overhead to convert > RDDs to Hive tables and back again. > Despite memory-level operations being where most of Spark's speed gains lie, > sometimes using disk-only may help with stability! -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[ https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14949076#comment-14949076 ] Glenn Strycker commented on SPARK-11004: Currently we could do the following from withing a linux script: 1) run Spark through creating the 2 RDDs, 2) save both to a Hive table, 3) end my Spark program, 4) run Hive Beeline on a "select * from table1 join table2 on columnname" and insert this into a new 3rd table, 5) start a new Spark program to continue on from the first program in step 3, loading the output of the Hive join into the results RDD. Rather than doing all of this overhead, it would be pretty cool if we could run the map and reduce jobs as a Hive join would do, only on our existing Spark RDDs, without needing to involve Hive or wrapper scripts, but doing everything from within a single Spark program. I believe the main stability gain is due to Hive performing everything on disk instead of memory. Since we already can checkpoint RDDs to memory, perhaps this ticket request could be accomplished by adding a feature to RDDs that would be performed on the checkpointed files, rather than the in-memory RDDs. We're just looking for a stability gain and ability to increase the potential size of RDDs in their operations. Right now our Hive Beeline scripts are out-performing Spark for these super massive table joins. > MapReduce Hive-like join operations for RDDs > > > Key: SPARK-11004 > URL: https://issues.apache.org/jira/browse/SPARK-11004 > Project: Spark > Issue Type: New Feature >Reporter: Glenn Strycker > > Could a feature be added to Spark that would use disk-only MapReduce > operations for the very largest RDD joins? > MapReduce is able to handle incredibly large table joins in a stable, > predictable way with gracious failures and recovery. I have applications > that are able to join 2 tables without error in Hive, but these same tables, > when converted into RDDs, are unable to join in Spark (I am using the same > cluster, and have played around with all of the memory configurations, > persisting to disk, checkpointing, etc., and the RDDs are just too big for > Spark on my cluster) > So, Spark is usually able to handle fairly large RDD joins, but occasionally > runs into problems when the tables are just too big (e.g. the notorious 2GB > shuffle limit issue, memory problems, etc.) There are so many parameters to > adjust (number of partitions, number of cores, memory per core, etc.) that it > is difficult to guarantee stability on a shared cluster (say, running Yarn) > with other jobs. > Could a feature be added to Spark that would use disk-only MapReduce commands > to do very large joins? > That is, instead of myRDD1.join(myRDD2), we would have a special operation > myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run > MapReduce, and then convert the results of the join back into a standard RDD. > This might add stability for Spark jobs that deal with extremely large data, > and enable developers to mix-and-match some Spark and MapReduce operations in > the same program, rather than writing Hive scripts and stringing together > Spark and MapReduce programs, which has extremely large overhead to convert > RDDs to Hive tables and back again. > Despite memory-level operations being where most of Spark's speed gains lie, > sometimes using disk-only may help with stability! -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[ https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14949171#comment-14949171 ] Glenn Strycker commented on SPARK-11004: So maybe we can simplify this idea down to forcing "disk-only shuffles" only in particular places. Spark could add a "force disk-only" parameter to the existing RDD join function so the command would look like this: rdd1.join(rdd2, diskonly = true) since it is the memory shuffles that seem to be causing my join problems. > MapReduce Hive-like join operations for RDDs > > > Key: SPARK-11004 > URL: https://issues.apache.org/jira/browse/SPARK-11004 > Project: Spark > Issue Type: New Feature >Reporter: Glenn Strycker > > Could a feature be added to Spark that would use disk-only MapReduce > operations for the very largest RDD joins? > MapReduce is able to handle incredibly large table joins in a stable, > predictable way with gracious failures and recovery. I have applications > that are able to join 2 tables without error in Hive, but these same tables, > when converted into RDDs, are unable to join in Spark (I am using the same > cluster, and have played around with all of the memory configurations, > persisting to disk, checkpointing, etc., and the RDDs are just too big for > Spark on my cluster) > So, Spark is usually able to handle fairly large RDD joins, but occasionally > runs into problems when the tables are just too big (e.g. the notorious 2GB > shuffle limit issue, memory problems, etc.) There are so many parameters to > adjust (number of partitions, number of cores, memory per core, etc.) that it > is difficult to guarantee stability on a shared cluster (say, running Yarn) > with other jobs. > Could a feature be added to Spark that would use disk-only MapReduce commands > to do very large joins? > That is, instead of myRDD1.join(myRDD2), we would have a special operation > myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run > MapReduce, and then convert the results of the join back into a standard RDD. > This might add stability for Spark jobs that deal with extremely large data, > and enable developers to mix-and-match some Spark and MapReduce operations in > the same program, rather than writing Hive scripts and stringing together > Spark and MapReduce programs, which has extremely large overhead to convert > RDDs to Hive tables and back again. > Despite memory-level operations being where most of Spark's speed gains lie, > sometimes using disk-only may help with stability! -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[
https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14949176#comment-14949176
]
Sean Owen commented on SPARK-11004:
---
I suppose I'd be surprised if using disk over memory helped, but you can easily
test this already with {{spark.shuffle.memoryFraction=0}} (or maybe it has to
be a very small value). Here are the configuration knobs:
http://spark.apache.org/docs/latest/configuration.html#shuffle-behavior
> MapReduce Hive-like join operations for RDDs
>
>
> Key: SPARK-11004
> URL: https://issues.apache.org/jira/browse/SPARK-11004
> Project: Spark
> Issue Type: New Feature
>Reporter: Glenn Strycker
>
> Could a feature be added to Spark that would use disk-only MapReduce
> operations for the very largest RDD joins?
> MapReduce is able to handle incredibly large table joins in a stable,
> predictable way with gracious failures and recovery. I have applications
> that are able to join 2 tables without error in Hive, but these same tables,
> when converted into RDDs, are unable to join in Spark (I am using the same
> cluster, and have played around with all of the memory configurations,
> persisting to disk, checkpointing, etc., and the RDDs are just too big for
> Spark on my cluster)
> So, Spark is usually able to handle fairly large RDD joins, but occasionally
> runs into problems when the tables are just too big (e.g. the notorious 2GB
> shuffle limit issue, memory problems, etc.) There are so many parameters to
> adjust (number of partitions, number of cores, memory per core, etc.) that it
> is difficult to guarantee stability on a shared cluster (say, running Yarn)
> with other jobs.
> Could a feature be added to Spark that would use disk-only MapReduce commands
> to do very large joins?
> That is, instead of myRDD1.join(myRDD2), we would have a special operation
> myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run
> MapReduce, and then convert the results of the join back into a standard RDD.
> This might add stability for Spark jobs that deal with extremely large data,
> and enable developers to mix-and-match some Spark and MapReduce operations in
> the same program, rather than writing Hive scripts and stringing together
> Spark and MapReduce programs, which has extremely large overhead to convert
> RDDs to Hive tables and back again.
> Despite memory-level operations being where most of Spark's speed gains lie,
> sometimes using disk-only may help with stability!
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
-
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[ https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14949187#comment-14949187 ] Glenn Strycker commented on SPARK-11004: Awesome -- thanks, I'll try that out. Is there a way to change this setting dynamically from within a Spark job, so that the fraction can be higher for most of the job and then drop down to 0 only for the difficult parts? > MapReduce Hive-like join operations for RDDs > > > Key: SPARK-11004 > URL: https://issues.apache.org/jira/browse/SPARK-11004 > Project: Spark > Issue Type: New Feature >Reporter: Glenn Strycker > > Could a feature be added to Spark that would use disk-only MapReduce > operations for the very largest RDD joins? > MapReduce is able to handle incredibly large table joins in a stable, > predictable way with gracious failures and recovery. I have applications > that are able to join 2 tables without error in Hive, but these same tables, > when converted into RDDs, are unable to join in Spark (I am using the same > cluster, and have played around with all of the memory configurations, > persisting to disk, checkpointing, etc., and the RDDs are just too big for > Spark on my cluster) > So, Spark is usually able to handle fairly large RDD joins, but occasionally > runs into problems when the tables are just too big (e.g. the notorious 2GB > shuffle limit issue, memory problems, etc.) There are so many parameters to > adjust (number of partitions, number of cores, memory per core, etc.) that it > is difficult to guarantee stability on a shared cluster (say, running Yarn) > with other jobs. > Could a feature be added to Spark that would use disk-only MapReduce commands > to do very large joins? > That is, instead of myRDD1.join(myRDD2), we would have a special operation > myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run > MapReduce, and then convert the results of the join back into a standard RDD. > This might add stability for Spark jobs that deal with extremely large data, > and enable developers to mix-and-match some Spark and MapReduce operations in > the same program, rather than writing Hive scripts and stringing together > Spark and MapReduce programs, which has extremely large overhead to convert > RDDs to Hive tables and back again. > Despite memory-level operations being where most of Spark's speed gains lie, > sometimes using disk-only may help with stability! -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Commented] (SPARK-11004) MapReduce Hive-like join operations for RDDs
[ https://issues.apache.org/jira/browse/SPARK-11004?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=14949235#comment-14949235 ] Sean Owen commented on SPARK-11004: --- Per job, no I don't think so. It's a setting on the Spark configuration which is fairly global. You might get away with shutting down the context and then reinitializing a new one, but I think that might be incompatible with what you're trying to test. > MapReduce Hive-like join operations for RDDs > > > Key: SPARK-11004 > URL: https://issues.apache.org/jira/browse/SPARK-11004 > Project: Spark > Issue Type: New Feature >Reporter: Glenn Strycker > > Could a feature be added to Spark that would use disk-only MapReduce > operations for the very largest RDD joins? > MapReduce is able to handle incredibly large table joins in a stable, > predictable way with gracious failures and recovery. I have applications > that are able to join 2 tables without error in Hive, but these same tables, > when converted into RDDs, are unable to join in Spark (I am using the same > cluster, and have played around with all of the memory configurations, > persisting to disk, checkpointing, etc., and the RDDs are just too big for > Spark on my cluster) > So, Spark is usually able to handle fairly large RDD joins, but occasionally > runs into problems when the tables are just too big (e.g. the notorious 2GB > shuffle limit issue, memory problems, etc.) There are so many parameters to > adjust (number of partitions, number of cores, memory per core, etc.) that it > is difficult to guarantee stability on a shared cluster (say, running Yarn) > with other jobs. > Could a feature be added to Spark that would use disk-only MapReduce commands > to do very large joins? > That is, instead of myRDD1.join(myRDD2), we would have a special operation > myRDD1.mapReduceJoin(myRDD2) that would checkpoint both RDDs to disk, run > MapReduce, and then convert the results of the join back into a standard RDD. > This might add stability for Spark jobs that deal with extremely large data, > and enable developers to mix-and-match some Spark and MapReduce operations in > the same program, rather than writing Hive scripts and stringing together > Spark and MapReduce programs, which has extremely large overhead to convert > RDDs to Hive tables and back again. > Despite memory-level operations being where most of Spark's speed gains lie, > sometimes using disk-only may help with stability! -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
