DataFrame distinct vs RDD distinct
Hi everyone, there seems to be different implementations of the "distinct" feature in DataFrames and RDD and some performance issue with the DataFrame distinct API. In RDD.scala : def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope { map(x => (x, null)).reduceByKey((x, y) => x, numPartitions).map(_._1) } And in DataFrame : case class Distinct(partial: Boolean, child: SparkPlan) extends UnaryNode { override def output: Seq[Attribute] = child.output override def requiredChildDistribution: Seq[Distribution] = if (partial) UnspecifiedDistribution :: Nil else ClusteredDistribution(child.output) :: Nil *override def execute(): RDD[Row] = {** child.execute().mapPartitions { iter =>** val hashSet = new scala.collection.mutable.HashSet[Row]()* * var currentRow: Row = null** while (iter.hasNext) {** currentRow = iter.next()** if (!hashSet.contains(currentRow)) {** hashSet.add(currentRow.copy())** }** }* * hashSet.iterator** }** }*} I can try to reproduce more clearly the performance issue, but do you have any insights into why we can't have the same distinct strategy between DataFrame and RDD ? Regards, Olivier.
Re: DataFrame distinct vs RDD distinct
In 1.5, we will most likely just rewrite distinct in SQL to either use the Aggregate operator which will benefit from all the Tungsten optimizations, or have a Tungsten version of distinct for SQL/DataFrame. On Thu, May 7, 2015 at 1:32 AM, Olivier Girardot < o.girar...@lateral-thoughts.com> wrote: > Hi everyone, > there seems to be different implementations of the "distinct" feature in > DataFrames and RDD and some performance issue with the DataFrame distinct > API. > > In RDD.scala : > > def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] > = > withScope { map(x => (x, null)).reduceByKey((x, y) => x, > numPartitions).map(_._1) } > And in DataFrame : > > > case class Distinct(partial: Boolean, child: SparkPlan) extends UnaryNode { > override def output: Seq[Attribute] = child.output override def > requiredChildDistribution: Seq[Distribution] = if (partial) > UnspecifiedDistribution :: Nil else ClusteredDistribution(child.output) :: > Nil *override def execute(): RDD[Row] = {** child.execute().mapPartitions { > iter =>** val hashSet = new scala.collection.mutable.HashSet[Row]()* * var > currentRow: Row = null** while (iter.hasNext) {** currentRow = > iter.next()** > if (!hashSet.contains(currentRow)) {** hashSet.add(currentRow.copy())** }** > }* * hashSet.iterator** }** }*} > > > > > I can try to reproduce more clearly the performance issue, but do you have > any insights into why we can't have the same distinct strategy between > DataFrame and RDD ? > > Regards, > > Olivier. >
Re: DataFrame distinct vs RDD distinct
Ok, but for the moment, this seems to be killing performances on some computations... I'll try to give you precise figures on this between rdd and dataframe. Olivier. Le jeu. 7 mai 2015 à 10:08, Reynold Xin a écrit : > In 1.5, we will most likely just rewrite distinct in SQL to either use the > Aggregate operator which will benefit from all the Tungsten optimizations, > or have a Tungsten version of distinct for SQL/DataFrame. > > On Thu, May 7, 2015 at 1:32 AM, Olivier Girardot < > o.girar...@lateral-thoughts.com> wrote: > >> Hi everyone, >> there seems to be different implementations of the "distinct" feature in >> DataFrames and RDD and some performance issue with the DataFrame distinct >> API. >> >> In RDD.scala : >> >> def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): >> RDD[T] = >> withScope { map(x => (x, null)).reduceByKey((x, y) => x, >> numPartitions).map(_._1) } >> And in DataFrame : >> >> >> case class Distinct(partial: Boolean, child: SparkPlan) extends UnaryNode >> { >> override def output: Seq[Attribute] = child.output override def >> requiredChildDistribution: Seq[Distribution] = if (partial) >> UnspecifiedDistribution :: Nil else ClusteredDistribution(child.output) :: >> > Nil *override def execute(): RDD[Row] = {** child.execute().mapPartitions { >> iter =>** val hashSet = new scala.collection.mutable.HashSet[Row]()* * var >> currentRow: Row = null** while (iter.hasNext) {** currentRow = >> iter.next()** >> if (!hashSet.contains(currentRow)) {** hashSet.add(currentRow.copy())** >> }** >> }* * hashSet.iterator** }** }*} > > >> >> >> >> >> I can try to reproduce more clearly the performance issue, but do you have >> any insights into why we can't have the same distinct strategy between >> DataFrame and RDD ? >> >> Regards, >> >> Olivier. >> >
Re: DataFrame distinct vs RDD distinct
I'd happily merge a PR that changes the distinct implementation to be more like Spark core, assuming it includes benchmarks that show better performance for both the "fits in memory case" and the "too big for memory case". On Thu, May 7, 2015 at 2:23 AM, Olivier Girardot < o.girar...@lateral-thoughts.com> wrote: > Ok, but for the moment, this seems to be killing performances on some > computations... > I'll try to give you precise figures on this between rdd and dataframe. > > Olivier. > > Le jeu. 7 mai 2015 à 10:08, Reynold Xin a écrit : > > > In 1.5, we will most likely just rewrite distinct in SQL to either use > the > > Aggregate operator which will benefit from all the Tungsten > optimizations, > > or have a Tungsten version of distinct for SQL/DataFrame. > > > > On Thu, May 7, 2015 at 1:32 AM, Olivier Girardot < > > o.girar...@lateral-thoughts.com> wrote: > > > >> Hi everyone, > >> there seems to be different implementations of the "distinct" feature in > >> DataFrames and RDD and some performance issue with the DataFrame > distinct > >> API. > >> > >> In RDD.scala : > >> > >> def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): > >> RDD[T] = > >> withScope { map(x => (x, null)).reduceByKey((x, y) => x, > >> numPartitions).map(_._1) } > >> And in DataFrame : > >> > >> > >> case class Distinct(partial: Boolean, child: SparkPlan) extends > UnaryNode > >> { > >> override def output: Seq[Attribute] = child.output override def > >> requiredChildDistribution: Seq[Distribution] = if (partial) > >> UnspecifiedDistribution :: Nil else ClusteredDistribution(child.output) > :: > >> > > Nil *override def execute(): RDD[Row] = {** > child.execute().mapPartitions { > >> iter =>** val hashSet = new scala.collection.mutable.HashSet[Row]()* * > var > >> currentRow: Row = null** while (iter.hasNext) {** currentRow = > >> iter.next()** > >> if (!hashSet.contains(currentRow)) {** hashSet.add(currentRow.copy())** > >> }** > >> }* * hashSet.iterator** }** }*} > > > > > >> > >> > >> > >> > >> I can try to reproduce more clearly the performance issue, but do you > have > >> any insights into why we can't have the same distinct strategy between > >> DataFrame and RDD ? > >> > >> Regards, > >> > >> Olivier. > >> > > >
Re: DataFrame distinct vs RDD distinct
I'll try to reproduce what has been reported to me first :) and I'll let you know. Thanks ! Le jeu. 7 mai 2015 à 21:16, Michael Armbrust a écrit : > I'd happily merge a PR that changes the distinct implementation to be more > like Spark core, assuming it includes benchmarks that show better > performance for both the "fits in memory case" and the "too big for memory > case". > > On Thu, May 7, 2015 at 2:23 AM, Olivier Girardot < > o.girar...@lateral-thoughts.com> wrote: > >> Ok, but for the moment, this seems to be killing performances on some >> computations... >> I'll try to give you precise figures on this between rdd and dataframe. >> >> Olivier. >> >> Le jeu. 7 mai 2015 à 10:08, Reynold Xin a écrit : >> >> > In 1.5, we will most likely just rewrite distinct in SQL to either use >> the >> > Aggregate operator which will benefit from all the Tungsten >> optimizations, >> > or have a Tungsten version of distinct for SQL/DataFrame. >> > >> > On Thu, May 7, 2015 at 1:32 AM, Olivier Girardot < >> > o.girar...@lateral-thoughts.com> wrote: >> > >> >> Hi everyone, >> >> there seems to be different implementations of the "distinct" feature >> in >> >> DataFrames and RDD and some performance issue with the DataFrame >> distinct >> >> API. >> >> >> >> In RDD.scala : >> >> >> >> def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): >> >> RDD[T] = >> >> withScope { map(x => (x, null)).reduceByKey((x, y) => x, >> >> numPartitions).map(_._1) } >> >> And in DataFrame : >> >> >> >> >> >> case class Distinct(partial: Boolean, child: SparkPlan) extends >> UnaryNode >> >> { >> >> override def output: Seq[Attribute] = child.output override def >> >> requiredChildDistribution: Seq[Distribution] = if (partial) >> >> UnspecifiedDistribution :: Nil else >> ClusteredDistribution(child.output) :: >> >> >> > Nil *override def execute(): RDD[Row] = {** >> child.execute().mapPartitions { >> >> iter =>** val hashSet = new scala.collection.mutable.HashSet[Row]()* * >> var >> >> currentRow: Row = null** while (iter.hasNext) {** currentRow = >> >> iter.next()** >> >> if (!hashSet.contains(currentRow)) {** hashSet.add(currentRow.copy())** >> >> }** >> >> }* * hashSet.iterator** }** }*} >> > >> > >> >> >> >> >> >> >> >> >> >> I can try to reproduce more clearly the performance issue, but do you >> have >> >> any insights into why we can't have the same distinct strategy between >> >> DataFrame and RDD ? >> >> >> >> Regards, >> >> >> >> Olivier. >> >> >> > >> > >
RE: DataFrame distinct vs RDD distinct
Hi, Could you suggest alternative way of implementing distinct, e.g. via fold or aggregate? Both SQL distinct and RDD distinct fail on my dataset due to overflow of Spark shuffle disk. I have 7 nodes with 300GB dedicated to Spark shuffle each. My dataset is 2B rows, the field which I'm performing distinct has 23 distinct values. Best regards, Alexander -Original Message- From: Olivier Girardot [mailto:o.girar...@lateral-thoughts.com] Sent: Friday, May 08, 2015 12:50 PM To: Michael Armbrust; Olivier Girardot Cc: Reynold Xin; dev@spark.apache.org Subject: Re: DataFrame distinct vs RDD distinct I'll try to reproduce what has been reported to me first :) and I'll let you know. Thanks ! Le jeu. 7 mai 2015 à 21:16, Michael Armbrust a écrit : > I'd happily merge a PR that changes the distinct implementation to be > more like Spark core, assuming it includes benchmarks that show better > performance for both the "fits in memory case" and the "too big for > memory case". > > On Thu, May 7, 2015 at 2:23 AM, Olivier Girardot < > o.girar...@lateral-thoughts.com> wrote: > >> Ok, but for the moment, this seems to be killing performances on some >> computations... >> I'll try to give you precise figures on this between rdd and dataframe. >> >> Olivier. >> >> Le jeu. 7 mai 2015 à 10:08, Reynold Xin a écrit : >> >> > In 1.5, we will most likely just rewrite distinct in SQL to either >> > use >> the >> > Aggregate operator which will benefit from all the Tungsten >> optimizations, >> > or have a Tungsten version of distinct for SQL/DataFrame. >> > >> > On Thu, May 7, 2015 at 1:32 AM, Olivier Girardot < >> > o.girar...@lateral-thoughts.com> wrote: >> > >> >> Hi everyone, >> >> there seems to be different implementations of the "distinct" >> >> feature >> in >> >> DataFrames and RDD and some performance issue with the DataFrame >> distinct >> >> API. >> >> >> >> In RDD.scala : >> >> >> >> def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): >> >> RDD[T] = >> >> withScope { map(x => (x, null)).reduceByKey((x, y) => x, >> >> numPartitions).map(_._1) } >> >> And in DataFrame : >> >> >> >> >> >> case class Distinct(partial: Boolean, child: SparkPlan) extends >> UnaryNode >> >> { >> >> override def output: Seq[Attribute] = child.output override def >> >> requiredChildDistribution: Seq[Distribution] = if (partial) >> >> UnspecifiedDistribution :: Nil else >> ClusteredDistribution(child.output) :: >> >> >> > Nil *override def execute(): RDD[Row] = {** >> child.execute().mapPartitions { >> >> iter =>** val hashSet = new >> >> scala.collection.mutable.HashSet[Row]()* * >> var >> >> currentRow: Row = null** while (iter.hasNext) {** currentRow = >> >> iter.next()** >> >> if (!hashSet.contains(currentRow)) {** >> >> hashSet.add(currentRow.copy())** >> >> }** >> >> }* * hashSet.iterator** }** }*} >> > >> > >> >> >> >> >> >> >> >> >> >> I can try to reproduce more clearly the performance issue, but do >> >> you >> have >> >> any insights into why we can't have the same distinct strategy >> >> between DataFrame and RDD ? >> >> >> >> Regards, >> >> >> >> Olivier. >> >> >> > >> > >
RE: DataFrame distinct vs RDD distinct
The following worked for me as a workaround for distinct: val pf = sqlContext.parquetFile("hdfs://file") val distinctValuesOfColumn4 = pf.rdd.aggregate[scala.collection.mutable.HashSet[String]](new scala.collection.mutable.HashSet[String]())( (s, v) => s += v.getString(4), (s1, s2) => s1 ++= s2) Best regards, Alexander -Original Message- From: Ulanov, Alexander Sent: Monday, May 11, 2015 11:59 AM To: Olivier Girardot; Michael Armbrust Cc: Reynold Xin; dev@spark.apache.org Subject: RE: DataFrame distinct vs RDD distinct Hi, Could you suggest alternative way of implementing distinct, e.g. via fold or aggregate? Both SQL distinct and RDD distinct fail on my dataset due to overflow of Spark shuffle disk. I have 7 nodes with 300GB dedicated to Spark shuffle each. My dataset is 2B rows, the field which I'm performing distinct has 23 distinct values. Best regards, Alexander -Original Message- From: Olivier Girardot [mailto:o.girar...@lateral-thoughts.com] Sent: Friday, May 08, 2015 12:50 PM To: Michael Armbrust; Olivier Girardot Cc: Reynold Xin; dev@spark.apache.org Subject: Re: DataFrame distinct vs RDD distinct I'll try to reproduce what has been reported to me first :) and I'll let you know. Thanks ! Le jeu. 7 mai 2015 à 21:16, Michael Armbrust a écrit : > I'd happily merge a PR that changes the distinct implementation to be > more like Spark core, assuming it includes benchmarks that show better > performance for both the "fits in memory case" and the "too big for > memory case". > > On Thu, May 7, 2015 at 2:23 AM, Olivier Girardot < > o.girar...@lateral-thoughts.com> wrote: > >> Ok, but for the moment, this seems to be killing performances on some >> computations... >> I'll try to give you precise figures on this between rdd and dataframe. >> >> Olivier. >> >> Le jeu. 7 mai 2015 à 10:08, Reynold Xin a écrit : >> >> > In 1.5, we will most likely just rewrite distinct in SQL to either >> > use >> the >> > Aggregate operator which will benefit from all the Tungsten >> optimizations, >> > or have a Tungsten version of distinct for SQL/DataFrame. >> > >> > On Thu, May 7, 2015 at 1:32 AM, Olivier Girardot < >> > o.girar...@lateral-thoughts.com> wrote: >> > >> >> Hi everyone, >> >> there seems to be different implementations of the "distinct" >> >> feature >> in >> >> DataFrames and RDD and some performance issue with the DataFrame >> distinct >> >> API. >> >> >> >> In RDD.scala : >> >> >> >> def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): >> >> RDD[T] = >> >> withScope { map(x => (x, null)).reduceByKey((x, y) => x, >> >> numPartitions).map(_._1) } >> >> And in DataFrame : >> >> >> >> >> >> case class Distinct(partial: Boolean, child: SparkPlan) extends >> UnaryNode >> >> { >> >> override def output: Seq[Attribute] = child.output override def >> >> requiredChildDistribution: Seq[Distribution] = if (partial) >> >> UnspecifiedDistribution :: Nil else >> ClusteredDistribution(child.output) :: >> >> >> > Nil *override def execute(): RDD[Row] = {** >> child.execute().mapPartitions { >> >> iter =>** val hashSet = new >> >> scala.collection.mutable.HashSet[Row]()* * >> var >> >> currentRow: Row = null** while (iter.hasNext) {** currentRow = >> >> iter.next()** >> >> if (!hashSet.contains(currentRow)) {** >> >> hashSet.add(currentRow.copy())** >> >> }** >> >> }* * hashSet.iterator** }** }*} >> > >> > >> >> >> >> >> >> >> >> >> >> I can try to reproduce more clearly the performance issue, but do >> >> you >> have >> >> any insights into why we can't have the same distinct strategy >> >> between DataFrame and RDD ? >> >> >> >> Regards, >> >> >> >> Olivier. >> >> >> > >> > > B CB [ X ܚX KK[XZ[ ] ][ X ܚX P \ ˘\X K ܙ B ܈Y][ۘ[ [X[ K[XZ[ ] Z[ \ ˘\X K ܙ B - To unsubscribe, e-mail: dev-unsubscr...@spark.apache.org For additional commands, e-mail: dev-h...@spark.apache.org