Mailing-List: contact issues-help@spark.apache.org; run by ezmlm
Precedence: bulk
Date: Tue, 27 Sep 2016 04:32:20 +0000 (UTC)
From: "Herman van Hovell (JIRA)" <jira@apache.org>
To: issues@spark.apache.org
Message-ID: <JIRA.13007762.1474933357000.669996.1474950740540@Atlassian.JIRA>
In-Reply-To: <JIRA.13007762.1474933357000@Atlassian.JIRA>
References: <JIRA.13007762.1474933357000@Atlassian.JIRA> <JIRA.13007762.1474933357970@arcas>
Subject: [jira] [Commented] (SPARK-17673) Reused Exchange Aggregations
 Produce Incorrect Results
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Tue, 27 Sep 2016 04:32:23 -0000


    [ https://issues.apache.org/jira/browse/SPARK-17673?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15525036#comment-15525036 ] 

Herman van Hovell commented on SPARK-17673:
-------------------------------------------

Could you also share the optimized plan {{df.explain(true)}? I am wondering if the attributes are the same.

> Reused Exchange Aggregations Produce Incorrect Results
> ------------------------------------------------------
>
>                 Key: SPARK-17673
>                 URL: https://issues.apache.org/jira/browse/SPARK-17673
>             Project: Spark
>          Issue Type: Bug
>          Components: SQL
>    Affects Versions: 2.0.0, 2.0.1
>            Reporter: Russell Spitzer
>            Priority: Critical
>              Labels: correctness
>
> https://datastax-oss.atlassian.net/browse/SPARKC-429
> Was brought to my attention where the following code produces incorrect results
> {code}
>  val data = List(TestData("A", 1, 7))
>     val frame = session.sqlContext.createDataFrame(session.sparkContext.parallelize(data))
>     frame.createCassandraTable(
>       keySpaceName,
>       table,
>       partitionKeyColumns = Some(Seq("id")))
>     frame
>       .write
>       .format("org.apache.spark.sql.cassandra")
>       .mode(SaveMode.Append)
>       .options(Map("table" -> table, "keyspace" -> keySpaceName))
>       .save()
> val loaded = sparkSession.sqlContext
>   .read
>   .format("org.apache.spark.sql.cassandra")
>   .options(Map("table" -> table, "keyspace" -> ks))
>   .load()
>   .select("id", "col1", "col2")
> val min1 = loaded.groupBy("id").agg(min("col1").as("min"))
> val min2 = loaded.groupBy("id").agg(min("col2").as("min"))
>  min1.union(min2).show()
>     /* prints:
>       +---+---+
>       | id|min|
>       +---+---+
>       |  A|  1|
>       |  A|  1|
>       +---+---+
>      Should be 
>       | A| 1|
>       | A| 7|
>      */
> {code}
> I looked into the explain pattern and saw 
> {code}
> Union
> :- *HashAggregate(keys=[id#93], functions=[min(col1#94)])
> :  +- Exchange hashpartitioning(id#93, 200)
> :     +- *HashAggregate(keys=[id#93], functions=[partial_min(col1#94)])
> :        +- *Scan org.apache.spark.sql.cassandra.CassandraSourceRelation@7ec20844 [id#93,col1#94]
> +- *HashAggregate(keys=[id#93], functions=[min(col2#95)])
>    +- ReusedExchange [id#93, min#153], Exchange hashpartitioning(id#93, 200)
> {code}
> Which was different than using a parallelized collection as the DF backing. So I tested the same code with a Parquet backed DF and saw the same results.
> {code}
>     frame.write.parquet("garbagetest")
>     val parquet = sparkSession.read.parquet("garbagetest").select("id", "col1", "col2")
>     println("PDF")
>     parquetmin1.union(parquetmin2).explain()
>     parquetmin1.union(parquetmin2).show()
>     /* prints:
>       +---+---+
>       | id|min|
>       +---+---+
>       |  A|  1|
>       |  A|  1|
>       +---+---+
> */
> {code}
> Which leads me to believe there is something wrong with the reused exchange. 


--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org