[jira] [Comment Edited] (SPARK-18473) Correctness issue in INNER join result with window functions

["Herman van Hovell (JIRA)" <ji...@apache.org>]

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

Herman van Hovell edited comment on SPARK-18473 at 11/16/16 8:59 PM:
---------------------------------------------------------------------

This has been fixed in spark 2.0.2.

See the following notebook: https://databricks-prod-cloudfront.cloud.databricks.com/public/4027ec902e239c93eaaa8714f173bcfc/2728434780191932/3885409687420673/6987336228780374/latest.html


was (Author: hvanhovell):
This has been fixed in spark 2.0.2.

> Correctness issue in INNER join result with window functions
> ------------------------------------------------------------
>
>                 Key: SPARK-18473
>                 URL: https://issues.apache.org/jira/browse/SPARK-18473
>             Project: Spark
>          Issue Type: Bug
>          Components: PySpark, Spark Core, SQL
>    Affects Versions: 2.0.1
>            Reporter: peay
>
> I have stumbled onto a corner case where an INNER join appears to return incorrect results. I believe the join should behave as the identity, but instead, some values are shuffled around, and some are just plain wrong.
> This can be reproduced as follows: joining
> {code}
> +-----+---------+------+--------+--------+----------+------+
> |index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|sessId|
> +-----+---------+------+--------+--------+----------+------+
> |    1|        1|     1|       0|       1|         0|     1|
> |    2|        2|     0|       0|       1|         0|     1|
> |    1|        3|     1|       0|       2|         0|     2|
> +-----+---------+------+--------+--------+----------+------+
> {code}
> with
> {code}
> +------+
> |sessId|
> +------+
> |     1|
> |     2|
> +------+
> {code}
> The result is
> {code}
> +------+-----+---------+------+--------+--------+----------+
> |sessId|index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|
> +------+-----+---------+------+--------+--------+----------+
> |     1|    2|        2|     0|       0|       1|         0|
> |     2|    1|        1|     1|       0|       1|        -1|
> |     2|    1|        3|     1|       0|       2|         0|
> +------+-----+---------+------+--------+--------+----------+
> {code}
> Note how two rows have a sessId of 2 (instead of one row as expected), and how `fiftyCount` can now be negative while always zero in the original dataframe.
> The first dataframe uses two windows:
> - `hasOne` uses a `window.rowsBetween(-10, 0)`.
> - `hasFifty` uses a `window.rowsBetween(-10, -1)`.
> The result is *correct* if:
> - `hasFifty` is changed to `window.rowsBetween(-10, 0)` instead of  `window.rowsBetween(-10, -1)`.
> - I add {code}.fillna({ 'numOnesBefore': 0 }) {code} after the other call to `fillna` -- although there are no visible effect on the dataframe as shown by `show` as far as I can tell.
> - I use a LEFT OUTER join instead of INNER JOIN.
> - I write both dataframes to Parquet, read them back and join these.
> This can be reproduced in pyspark using:
> {code}
> import pyspark.sql.functions as F
> from pyspark.sql.functions import col
> from pyspark.sql.window import Window
> df1 = sql_context.createDataFrame(
>     pd.DataFrame({"index": [1, 2, 1], "timeStamp": [1, 2, 3]})
> )
> window = Window.partitionBy(F.lit(1)).orderBy("timeStamp", "index")
> df2 = (
>     df1
>     .withColumn("hasOne", (col("index") == 1).cast("int"))
>     .withColumn("hasFifty", (col("index") == 50).cast("int"))
>     .withColumn("numOnesBefore", F.sum(col("hasOne")).over(window.rowsBetween(-10, 0)))
>     .withColumn("numFiftyStrictlyBefore", F.sum(col("hasFifty")).over(window.rowsBetween(-10, -1)))
>     .fillna({ 'numFiftyStrictlyBefore': 0 })
>     .withColumn("sessId", col("numOnesBefore") - col("numFiftyStrictlyBefore"))
> )
> df_selector = sql_context.createDataFrame(pd.DataFrame({"sessId": [1, 2]}))
> df_joined = df_selector.join(df2, "sessId", how="inner")
> df2.show()
> df_selector.show()
> df_joined.show()
> {code}



--
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