You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by jagaximo <ta...@dwango.co.jp> on 2015/01/20 04:38:10 UTC
How to compute RDD[(String, Set[String])] that include large Set
i want compute RDD[(String, Set[String])] that include a part of large size
’Set[String]’.
--------------
val hoge: RDD[(String, Set[String])] = ...
val reduced = hoge.reduceByKey(_ ++ _) //<= create large size Set (shuffle
read size 7GB)
val counted = reduced.map{ case (key, strSeq) => s”$key\t${strSeq.size}"}
counted.saveAsText(“/path/to/save/dir")
----------
Look Spark UI, In stage of saveAsText, lost executor and starting resubmit.
then spark continue much lost executor.
i think, approach for this problem solving, make ‘RDD[(String,
RDD[String])]’ , union RDD[String], and distinct count. but create RDD in
RDD, NullPointerException has occured. maybe impossible this operation
What might be the issue and possible solution?
please lend your wisdom
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-that-include-large-Set-tp21248.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: How to compute RDD[(String, Set[String])] that include large Set
Posted by "Kevin (Sangwoo) Kim" <ke...@apache.org>.
If keys are not too many,
You can do like this:
val data = List(
("A", Set(1,2,3)),
("A", Set(1,2,4)),
("B", Set(1,2,3))
)
val rdd = sc.parallelize(data)
rdd.persist()
rdd.filter(_._1 == "A").flatMap(_._2).distinct.count
rdd.filter(_._1 == "B").flatMap(_._2).distinct.count
rdd.unpersist()
==
data: List[(String, scala.collection.mutable.Set[Int])] = List((A,Set(1, 2,
3)), (A,Set(1, 2, 4)), (B,Set(1, 2, 3))) rdd:
org.apache.spark.rdd.RDD[(String, scala.collection.mutable.Set[Int])] =
ParallelCollectionRDD[6940] at parallelize at <console>:66 res332: rdd.type
= ParallelCollectionRDD[6940] at parallelize at <console>:66 res334: Long =
4 res335: Long = 3 res336: rdd.type = ParallelCollectionRDD[6940] at
parallelize at <console>:66
Regards,
Kevin
On Tue Jan 20 2015 at 2:53:22 PM jagaximo <ta...@dwango.co.jp>
wrote:
> That i want to do, get unique count for each key. so take map() or
> countByKey(), not get unique count. (because duplicate string is likely to
> be counted)...
>
>
>
>
> --
> View this message in context: http://apache-spark-user-list.
> 1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-
> that-include-large-Set-tp21248p21254.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>
Re: How to compute RDD[(String, Set[String])] that include large
Set
Posted by jagaximo <ta...@dwango.co.jp>.
That i want to do, get unique count for each key. so take map() or
countByKey(), not get unique count. (because duplicate string is likely to
be counted)...
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-that-include-large-Set-tp21248p21254.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: How to compute RDD[(String, Set[String])] that include large Set
Posted by "Kevin (Sangwoo) Kim" <ke...@apache.org>.
In your code, you're doing combination of large sets, like
(set1 ++ set2).size
which is not a good idea.
(rdd1 ++ rdd2).distinct
is equivalent implementation and will compute in distributed manner.
Not very sure your computation on key'd sets are feasible to be transformed
into RDDs.
Regards,
Kevin
On Tue Jan 20 2015 at 1:57:52 PM Kevin Jung <it...@samsung.com> wrote:
> As far as I know, the tasks before calling saveAsText are transformations
> so
> that they are lazy computed. Then saveAsText action performs all
> transformations and your Set[String] grows up at this time. It creates
> large
> collection if you have few keys and this causes OOM easily when your
> executor memory and fraction settings are not suitable for computing this.
> If you want only collection counts by keys , you can use countByKey() or
> map() RDD[(String, Set[String])] to RDD[(String,Long)] after creating hoge
> RDD to make reduceByKey collect only counts of keys.
>
>
>
> --
> View this message in context: http://apache-spark-user-list.
> 1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-
> that-include-large-Set-tp21248p21251.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>
Re: How to compute RDD[(String, Set[String])] that include large
Set
Posted by Kevin Jung <it...@samsung.com>.
As far as I know, the tasks before calling saveAsText are transformations so
that they are lazy computed. Then saveAsText action performs all
transformations and your Set[String] grows up at this time. It creates large
collection if you have few keys and this causes OOM easily when your
executor memory and fraction settings are not suitable for computing this.
If you want only collection counts by keys , you can use countByKey() or
map() RDD[(String, Set[String])] to RDD[(String,Long)] after creating hoge
RDD to make reduceByKey collect only counts of keys.
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-that-include-large-Set-tp21248p21251.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: How to compute RDD[(String, Set[String])] that include large Set
Posted by "Kevin (Sangwoo) Kim" <ke...@apache.org>.
Great to hear you got solution!!
Cheers!
Kevin
On Wed Jan 21 2015 at 11:13:44 AM jagaximo <ta...@dwango.co.jp>
wrote:
> Kevin (Sangwoo) Kim wrote
> > If keys are not too many,
> > You can do like this:
> >
> > val data = List(
> > ("A", Set(1,2,3)),
> > ("A", Set(1,2,4)),
> > ("B", Set(1,2,3))
> > )
> > val rdd = sc.parallelize(data)
> > rdd.persist()
> >
> > rdd.filter(_._1 == "A").flatMap(_._2).distinct.count
> > rdd.filter(_._1 == "B").flatMap(_._2).distinct.count
> > rdd.unpersist()
> >
> > ==
> > data: List[(String, scala.collection.mutable.Set[Int])] = List((A,Set(1,
> > 2, 3)), (A,Set(1, 2, 4)), (B,Set(1, 2, 3)))
> > rdd: org.apache.spark.rdd.RDD[(String, scala.collection.mutable.Set[
> Int])]
> > = ParallelCollectionRDD[6940] at parallelize at
> > <console>
> > :66
> > res332: rdd.type = ParallelCollectionRDD[6940] at parallelize at
> > <console>
> > :66
> > res334: Long = 4
> > res335: Long = 3
> > res336: rdd.type = ParallelCollectionRDD[6940] at parallelize at
> > <console>
> > :66
> >
> > Regards,
> > Kevin
>
> Wow, Got it! good solution
> Fortunately, I know what keys have large size Set, I was able to adopt this
> approach.
>
> thanks!
>
>
>
>
> --
> View this message in context: http://apache-spark-user-list.
> 1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-
> that-include-large-Set-tp21248p21275.html
> Sent from the Apache Spark User List mailing list archive at Nabble.com.
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
> For additional commands, e-mail: user-help@spark.apache.org
>
>
Re: How to compute RDD[(String, Set[String])] that include large
Set
Posted by jagaximo <ta...@dwango.co.jp>.
Kevin (Sangwoo) Kim wrote
> If keys are not too many,
> You can do like this:
>
> val data = List(
> ("A", Set(1,2,3)),
> ("A", Set(1,2,4)),
> ("B", Set(1,2,3))
> )
> val rdd = sc.parallelize(data)
> rdd.persist()
>
> rdd.filter(_._1 == "A").flatMap(_._2).distinct.count
> rdd.filter(_._1 == "B").flatMap(_._2).distinct.count
> rdd.unpersist()
>
> ==
> data: List[(String, scala.collection.mutable.Set[Int])] = List((A,Set(1,
> 2, 3)), (A,Set(1, 2, 4)), (B,Set(1, 2, 3)))
> rdd: org.apache.spark.rdd.RDD[(String, scala.collection.mutable.Set[Int])]
> = ParallelCollectionRDD[6940] at parallelize at
> <console>
> :66
> res332: rdd.type = ParallelCollectionRDD[6940] at parallelize at
> <console>
> :66
> res334: Long = 4
> res335: Long = 3
> res336: rdd.type = ParallelCollectionRDD[6940] at parallelize at
> <console>
> :66
>
> Regards,
> Kevin
Wow, Got it! good solution
Fortunately, I know what keys have large size Set, I was able to adopt this
approach.
thanks!
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-that-include-large-Set-tp21248p21275.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org
Re: How to compute RDD[(String, Set[String])] that include large
Set
Posted by Pankaj Narang <pa...@gmail.com>.
Instead of counted.saveAsText(“/path/to/save/dir") if you call
counted.collect what happens ?
If you still face the same issue please paste the stacktrace here.
--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/How-to-compute-RDD-String-Set-String-that-include-large-Set-tp21248p21250.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.
---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org